将字典列表分解为 Spark 中的其他列

Question

我目前有一个 UDF，它采用一列 xml 字符串并将其解析为字典列表。然后，我想根据键值对将该字典列列表分解为其他列。

输入看起来像这样：

   id  type  length  parsed    
0  1   A     144     [{'key1':'value1'},{'key1':'value2', 'key2':'value3'},...]
1  1   B     20      [{'key1':'value4'},{'key2':'value5'},...]
2  4   A     54      [{'key3':'value6'},...]

我希望输出看起来像这样：

   id  type  length  key1             key2     key3
0  1   A     144     [value1,value2]  value3
1  1   B     20      value4           value5
2  4   A     54                                value6

我已经能够像这样在 Pandas 中做到这一点：

s = data['parsed xml'].explode()
df_join = (pd.DataFrame(s.tolist(), index = s.index)
             .stack()
             .groupby(level=[0,1])
             .agg(list)
             .apply(lambda x: x[0] if len(x)==1 else x)
             .unstack(fill_value='')
          )
t = data.join(df_join, lsuffix = '_x', rsuffix = '_y')

问题是我无法在 Spark 中转换此 Pandas 代码（我无法使用 Pandas），这会给我相同的结果。

我将使用的 Spark 是 1.6.0。

Answer 1

您可以使用explode 两次来执行此操作 - 一次是分解数组，一次是分解数组的地图元素。此后，您可以将pivot 与collect_list 聚合一起使用。

from pyspark.sql.functions import explode,collect_list
#explode array
df_1 = df.withColumn('exploded_arr',explode('parsed')) 
#explode maps of array elements
df_2 = df_1.select(*df_1.columns,explode('exploded_arr')) #the default column names returned after exploding a map are `key`,`value`. change them as needed
#pivot with aggregation
df_2.groupBy("id","length","type").pivot("key").agg(collect_list("value")).show()