PySpark实现Map列值行级求和并生成新列的问题

你的问题出在累加器初始值的类型不匹配：原代码里用0（整数类型）作为aggregate的初始值，但Map中的数值是浮点数，导致类型冲突报错。只需把初始值改成浮点型的0.0即可解决。

正确实现代码 #

方式1：修正SQL表达式的初始值类型 #

import pyspark.sql.functions as F

df = df.withColumn('total', F.expr('aggregate(map_values(col), 0.0, (acc, x) -> acc + x)'))

方式2：纯PySpark函数写法 #

import pyspark.sql.functions as F
from pyspark.sql.types import DoubleType

df = df.withColumn(
    'total',
    F.aggregate(
        F.map_values('col'),
        F.lit(0.0).cast(DoubleType()),
        lambda acc, x: acc + x
    )
)

验证结果 #

运行后DataFrame会得到预期输出：

ID           col                                                total
1            [item1 -> 0.2, Item2 -> 0.3, item3 -> 0.4]          0.9
2            [item2 -> 0.1, Item2 -> 0.7, item3 -> 0.2]          1.0

内容的提问来源于stack exchange，提问作者Abhishek Patil