PySpark分组合并DataFrame多行name字段问题求助

我来帮你梳理下问题，再给出高效的解决办法：

首先你代码里的agg用法是错误的——PySpark的groupBy.agg()并不支持直接传入列名加Python lambda的写法，因为Spark需要的是可序列化的原生聚合函数，普通lambda无法被Spark的分布式执行计划识别，也不符合API的参数要求。

下面是两种基于Spark原生函数的解决方案，性能比自定义UDF好很多：

方法1：保留重复值的拼接（推荐） #

先用collect_list把同category下的name收集成列表，再用concat_ws以逗号加空格为分隔符拼接成字符串：

from pyspark.sql import functions as F

# 分组聚合并拼接name列
result_df = spark_df.groupBy("category").agg(
    F.concat_ws(", ", F.collect_list("name")).alias("name")
)

# 查看结果
result_df.show()

执行后会得到你想要的输出：

+---------+-------+
|category|   name|
+---------+-------+
|        A|A1, A2|
|        B|B1, B2|
+---------+-------+

方法2：自动去重的拼接 #

如果你的数据里存在重复的name，想要去重后再拼接，可以用collect_set代替collect_list：

result_df = spark_df.groupBy("category").agg(
    F.concat_ws(", ", F.collect_set("name")).alias("name")
)

原代码失效的原因 #

你写的spark_df.groupby('category').agg('name', lambda x:x + ', ')不符合PySpark的API规范：

• agg()的参数可以是字典（比如{"name": "collect_list"}），或者多个Column对象（比如F.collect_list("name")）
• 直接传入Python lambda的话，Spark无法将这个函数转化为分布式执行的逻辑，就算封装成UDF，性能也远不如原生聚合函数，完全没必要这么做。

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