如何在PySpark中实现类似Pandas的多列透视操作

PySpark原生pivot方法仅支持单列透视，无法直接像Pandas那样传入多列作为透视维度。要实现你需要的效果，核心思路是先将多列透视维度拼接成单个复合列，再基于该列完成透视，最后调整列名匹配预期格式。

完整实现代码 #

from pyspark.sql import functions as F

# 1. 拼接透视维度为复合列（选数据中不存在的分隔符避免冲突）
df_with_pivot_col = df_d1.withColumn(
    "pivot_col",
    F.concat_ws(
        "__",  # 用双下划线做分隔符，规避字段值里的普通下划线冲突
        F.col("New/Rebid"),
        F.col("Year").cast("string"),
        F.col("Deal Conclusion")
    )
)

# 2. 分组+透视+聚合
df_pivoted = df_with_pivot_col.groupBy(
    "GFCID", "GFCID Name", "Client Priority"
).pivot("pivot_col").agg(
    F.count("Deal").alias("Deal"),
    F.sum("Revenue").alias("Revenue"),
    F.collect_set("Location").alias("Location"),
    F.collect_set("Deal Location").alias("Deal Location")
)

# 3. 调整列名（模拟Pandas多级列结构，用下划线分隔维度）
final_df = df_pivoted
for col_name in df_pivoted.columns:
    if "__" in col_name:
        # 拆分复合列的三个维度
        new_rebid, year, deal_conclusion = col_name.split("__")[:3]
        # 提取聚合字段名
        agg_field = col_name.split(".")[-1].split("__")[-1]
        # 生成新列名：聚合字段_New/Rebid_Year_Deal Conclusion
        new_col_name = f"{agg_field}_{new_rebid}_{year}_{deal_conclusion}"
        final_df = final_df.withColumnRenamed(col_name, new_col_name)

# 查看结果
final_df.show(truncate=False)

关键细节说明 #

• 复合列拼接：用concat_ws将New/Rebid、Year（转字符串）、Deal Conclusion拼接成单列，双下划线分隔符可根据实际数据替换为其他无冲突字符。
• 聚合函数映射：
  • Pandas的Deal: count → PySpark用F.count("Deal")
  • Pandas的Revenue: sum → PySpark用F.sum("Revenue")
  • Pandas的Location: lambda x: set(x) → PySpark用F.collect_set("Location")（自动去重，与set逻辑完全一致）
• 列名调整：PySpark不支持真正的多级列，因此用下划线分隔维度的方式模拟Pandas的多级列结构，方便后续识别维度信息。
• 空值处理：透视后无匹配数据的位置自动填充null，与Pandas的NaN行为一致。

原有代码报错原因 #

• PySpark的pivot方法仅接受单个列名作为参数，无法直接传入多列，因此.pivot('New/Rebid','Year','Deal Conclusion')会触发语法错误。
• 之前的agg中使用F.first('Year')逻辑错误：Year是透视维度，不需要对其做聚合，而是要作为透视的组成部分。

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