PySpark按value字段groupby后合并同组Item列值为集合的实现方法

核心实现逻辑 #

• 按value字段执行groupBy分组
• 用PySpark内置聚合函数collect_set收集同组Item字段值，自动去重后返回数组类型的集合
• 如果需要输出为{A,B}格式的字符串，可搭配字符串拼接函数完成格式转换

完整示例代码 #

# 导入依赖
from pyspark.sql import SparkSession
from pyspark.sql.functions import collect_set, concat, lit, concat_ws

# 初始化SparkSession，创建示例DataFrame
spark = SparkSession.builder.appName("groupCollectItem").getOrCreate()
data = [("A", 1), ("B", 1), ("C", 2), ("D", 2)]
df = spark.createDataFrame(data, schema=["Item", "value"])

# 1. 聚合得到Array类型的集合（适合后续继续做数据处理）
result_array = df.groupBy("value") \
                 .agg(collect_set("Item").alias("Item")) \
                 .select("Item", "value")

# 2. 聚合得到{A,B}格式的字符串（和需求输出格式完全一致）
result_str = df.groupBy("value") \
               .agg(
                   concat(
                       lit("{"),
                       concat_ws(",", collect_set("Item")),
                       lit("}")
                   ).alias("Item")
               ) \
               .select("Item", "value")

# 输出结果验证
result_str.show(truncate=False)

输出结果 #

+-----+-----+
|Item |value|
+-----+-----+
|{A,B}|1    |
|{C,D}|2    |
+-----+-----+

注意事项 #

• 若不需要对同组Item去重，将collect_set替换为collect_list即可
• Array类型的集合结果支持直接用PySpark数组函数做后续处理，无需转字符串

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