我使用下面的函数来分解一个深度嵌套的json(具有嵌套的结构和数组)。
# Flatten nested df
def flatten_df(nested_df):
for col in nested_df.columns:
array_cols = [c[0] for c in nested_df.dtypes if c[1][:5] == 'array']
for col in array_cols:
nested_df =nested_df.withColumn(col, F.explode_outer(nested_df[col]))
nested_cols = [c[0] for c in nested_df.dtypes if c[1][:6] == 'struct']
if len(nested_cols) == 0:
return nested_df
flat_cols = [c[0] for c in nested_df.dtypes if c[1][:6] != 'struct']
flat_df = nested_df.select(flat_cols +
[F.col(nc+'.'+c).alias(nc+'_'+c)
for nc in nested_cols
for c in nested_df.select(nc+'.*').columns])
return flatten_df(flat_df)
我成功地爆炸了。但是我还想在分解的Dataframe中添加元素的顺序或索引。所以在上面的代码中我替换 explode_outer
函数到 posexplode_outer
. 但我得到下面的错误
An error was encountered:
'The number of aliases supplied in the AS clause does not match the number of columns output by the UDTF expected 2 aliases'
我试着换衣服 nested_df.withColumn
至 nested_df.select
但我没有成功。有谁能帮我分解嵌套的json,但同时保持数组元素在分解的dataframe中作为列的顺序。
4条答案
按热度按时间uqjltbpv1#
以dataframe格式读取json数据并创建视图或表。在sparksql中,可以使用使用别名引用的number-of-laterviewexplode方法。如果json数据结构是struct类型,那么可以使用点来表示结构。级别1.2
nhaq1z212#
错误是因为posexplode\u outer返回两列pos和col,所以不能与column()一起使用。这可以在选择中使用,如下面的代码所示
结果:
如果需要重命名列,可以使用.withcolumnrenamed()函数
wpcxdonn3#
替换
nested_df =nested_df.withColumn(col, F.explode_outer(nested_df[col]))
与nested_df = df.selectExpr("*", f"posexplode({col}) as (position,col)").drop(col)
您可能需要编写一些逻辑来将列名替换为original,但应该很简单ddhy6vgd4#
您可以尝试使用列表理解选择来分解现有代码中的arraytype列:
例子: