PySpark将DataFrame按用户分组转为固定3组ticker、date列的实现方法

解法思路 #

你之前的方案出问题是因为 pivot 的对象是ticker字段，自然会生成和不同ticker数量一致的列。你只需要先给每个user内部的行生成1~3的序号，再对序号做pivot，就能得到固定3组的列。 #

完整代码示例 #

# 导入依赖
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 1. 给每个用户的记录按顺序编号，业务固定3行所以序号只会是1、2、3
# 如果需要固定排序规则，把orderBy的参数改为对应字段即可，比如orderBy("ticker")
w = Window.partitionBy("user").orderBy(F.lit(1))
df_with_rn = df.withColumn("row_num", F.row_number().over(w))

# 2. 按用户分组，对序号做pivot，聚合取对应字段值
# 聚合用first是因为每个user+row_num的组合唯一，任意聚合函数都可以正常返回结果
pivoted_df = df_with_rn.groupBy("user") \
    .pivot("row_num") \
    .agg(
        F.first("ticker").alias("ticker"),
        F.first("date").alias("date")
    )

# 3. 重命名列为要求的格式（默认生成的列名为`1_ticker`、`1_date`，转换为`ticker_1`、`date_1`）
final_cols = ["user"]
for num in range(1,4):
    final_cols.append(f"{num}_ticker as ticker_{num}")
    final_cols.append(f"{num}_date as date_{num}")

final_df = pivoted_df.selectExpr(final_cols)

特殊场景说明 #

如果你实际需求是将多个用户的记录合并为同一行（比如你示例中u2/u3/u4各一条合并为一行），只需要调整窗口partitionBy的分组规则，新增自定义的分组字段把需要合并的用户归为同一组即可，核心逻辑不需要改动。

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