Spark/Python Dataframe行转列：将AEMainCategoryKey转为列并合并行

问题背景 #

现有样本数据包含ProductionDate、CostCenterKey、AEMainCategoryKey、AELossMagnitude字段，需要将AEMainCategoryKey的取值转为独立列，使ProductionDate与CostCenterKey的每组组合对应一行，缺失值填充为0。但使用Pandas的pivot_table代码后未得到预期结果。

问题原因 #

1. 未将pivot_table的运算结果赋值回变量，原pandas_df未被修改，display的仍是转换前的原始数据。
2. 默认聚合逻辑可能不符合需求（pivot_table默认用mean聚合，若同一组合有多条数据会计算均值，需根据需求显式指定）。
3. 透视后的结果保留了多级索引，未转为普通列，可能不符合预期格式。

修正方案 #

方案1：修正Pandas代码 #

from pyspark.sql import SparkSession
import pandas as pd

# 创建Spark DataFrame
spark_df = sqlContext.sql("select * from hive_metastore.asseteffectiveness.asset_effectiveness_maincat where productiondate in ('2022-01-01','2022-01-02') and costcenterkey in (100030,100040)")
# 转换为Pandas DataFrame
pandas_df = spark_df.toPandas()

# 执行透视并赋值回变量，指定聚合方式（如sum，根据实际需求调整），重置索引
pandas_df = pandas_df.pivot_table(
    index=['ProductionDate','CostCenterKey'],
    columns=['AEMainCategoryKey'],
    values='AELossMagnitude',
    fill_value=0,
    aggfunc='sum'  # 若同一组合有多条数据，指定聚合逻辑，可选mean/sum等
).reset_index()

# 重置列名，去掉多级列名的层级
pandas_df.columns.name = None

display(pandas_df)

方案2：使用Spark原生Pivot（推荐大数据场景） #

无需转换为Pandas，直接用Spark的pivot操作更高效，避免内存压力：

from pyspark.sql import functions as F

# 创建Spark DataFrame
spark_df = sqlContext.sql("select * from hive_metastore.asseteffectiveness.asset_effectiveness_maincat where productiondate in ('2022-01-01','2022-01-02') and costcenterkey in (100030,100040)")

# 执行Spark透视：分组后透视，填充缺失值为0
result_df = spark_df.groupBy('ProductionDate', 'CostCenterKey') \
    .pivot('AEMainCategoryKey') \
    .agg(F.coalesce(F.sum('AELossMagnitude'), F.lit(0)))

display(result_df)

关键说明 #

• 若同一ProductionDate+CostCenterKey+AEMainCategoryKey组合有多条数据，需明确聚合逻辑（求和/均值等），否则Pandas默认用均值，Spark默认取第一个值。
• 重置索引和列名是为了让结果符合“每行对应一组组合，所有字段为普通列”的预期格式。

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