PySpark分组后扁平化实现：多列DataFrame按Col1分组生成Col2列表

我来帮你搞定这个需求！咱们把整个流程拆成两步，同时兼顾DataFrame里的其他列，一步步来实现。

核心思路 #

咱们的需求本质是两个操作的组合：

1. 按Col1分组，把每组的Col2值收集成一个列表
2. 将生成的列表“拆”成单独的行（扁平化），同时保留所有其他列

具体实现步骤 #

1. 分组聚合生成Col2列表 #

首先用groupBy指定分组键，再用collect_list函数把每组的Col2值打包成列表。这里关键是处理其他列：

• 如果其他列和Col1是一一对应的（比如Col1是用户ID，其他列是用户的固定属性），直接把这些列也加到groupBy里即可；
• 如果其他列在同一Col1下有多个值，得先明确业务逻辑，比如用first()/last()取分组内的某个值，不然会报错。

举个实际例子，假设我们的DataFrame有Col1、Col2、Col3、Col4几列：

from pyspark.sql import functions as F

# 创建示例DataFrame
df = spark.createDataFrame(
    [("A", "X", "user_a_age", "user_a_city"),
     ("A", "Y", "user_a_age", "user_a_city"),
     ("B", "Z", "user_b_age", "user_b_city")],
    ["Col1", "Col2", "Col3", "Col4"]
)

# 分组聚合：因为Col3、Col4和Col1一一对应，所以一起加入groupBy
grouped_df = df.groupBy("Col1", "Col3", "Col4").agg(F.collect_list("Col2").alias("Col2_list"))

2. 扁平化列表（拆分行） #

接下来用explode函数把Col2_list里的每个元素拆成单独的行，最后删掉临时的列表列就搞定了：

# 扁平化列表，还原Col2列
flattened_df = grouped_df.withColumn("Col2", F.explode("Col2_list")).drop("Col2_list")

最终效果 #

运行完上面的代码后，flattened_df就会回到和原始DataFrame类似的结构，但完成了分组-聚合-扁平化的完整流程（如果原始数据有重复值，会先聚合去重再展开，根据你的需求调整即可）。

额外注意事项 #

• 如果需要生成无重复值的列表，可以把collect_list换成collect_set，但要注意集合是无序的；
• 如果其他列和Col1不是一一对应，比如同一Col1下有不同的Col3值，你需要先确定业务规则：是保留所有组合，还是取某个特定值？比如用F.first("Col3").alias("Col3")来取分组内第一个Col3的值。

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