将地图数组组合成 pyspark 数据框中的单个地图答案

【问题标题】：Combine array of maps into single map in pyspark dataframe将地图数组组合成 pyspark 数据框中的单个地图
【发布时间】：2021-01-31 07:35:45
【问题描述】：

是否有类似于 collect_list 或 collect_set 的函数将一列地图聚合到（分组的）pyspark 数据帧中的单个地图中？例如，此函数可能具有以下行为：

>>>df.show()

+--+---------------------------------+
|id|                             map |
+--+---------------------------------+
| 1|                    Map(k1 -> v1)|
| 1|                    Map(k2 -> v2)|
| 1|                    Map(k3 -> v3)|
| 2|                    Map(k5 -> v5)|
| 3|                    Map(k6 -> v6)|
| 3|                    Map(k7 -> v7)|
+--+---------------------------------+

>>>df.groupBy('id').agg(collect_map('map')).show()

+--+----------------------------------+
|id|                 collect_map(map) |
+--+----------------------------------+
| 1| Map(k1 -> v1, k2 -> v2, k3 -> v3)|
| 2|                     Map(k5 -> v5)|
| 3|           Map(k6 -> v6, k7 -> v7)|
+--+----------------------------------+

使用其他 collect_ 聚合之一和 udf 产生所需结果可能不会太难，但似乎这样的东西应该已经存在。

【问题讨论】：

标签： pyspark spark-dataframe pyspark-sql

【解决方案1】：

我知道在其他人有机会回答之前为您自己的问题提供答案可能是一种糟糕的形式，但如果有人正在寻找基于 udf 的版本，这里有一个可能的答案。

from pyspark.sql.functions import udf,collect_list
from pyspark.sql.types import MapType,StringType

combineMap=udf(lambda maps: {key:f[key] for f in maps for key in f},
               MapType(StringType(),StringType()))

df.groupBy('id')\
  .agg(collect_list('map')\
  .alias('maps'))\
  .select('id',combineMap('maps').alias('combined_map')).show()

【讨论】：

【解决方案2】：

建议的 concat_map 解决方案不起作用，并且此解决方案不使用 UDF。
对于火花>=2.4

(df
.groupBy(f.col('id'))
.agg(f.collect_list(f.col('map')).alias('maps'), 
.select('id',
        f.expr('aggregate(slice(maps, 2, size(maps)), maps[0], (acc, element) -> map_concat(acc, element))').alias('mapsConcatenated')
        )
)

collect_list 会忽略空值，因此在聚合函数中使用 map_concat 时无需担心它们。

【讨论】：

【解决方案3】：

它是 pyspark 版本中的 map_concat >= 2.4

【讨论】：

我尝试使用 map_concat 作为聚合器，但出现此错误：“表达式 'test2' 既不在 group by 中，也不是聚合函数。” test2 是包含地图的列的名称。