本文介绍如何使用 PySpark 的 pivot() 方法，将百万级长格式数据按 object_id 聚合并动态转为宽表结构，适用于属性维度固定（如 ≤20 个）的 ETL 场景。
本文介绍如何使用 pyspark 的 `pivot()` 方法，将百万级长格式数据按 `object_id` 聚合并动态转为宽表结构，适用于属性维度固定（如 ≤20 个）的 etl 场景。
在大数据处理中，常需将“键值对”形式的长表（EAV 模型）转换为便于分析的宽表结构——例如将每个 object_id 对应的多个属性（name、age、number 等）展开为独立列。当数据量达百万行以上、但属性种类有限（如本例中 attr_name 不超过 20 个），PySpark 提供的 pivot() 是简洁高效的解决方案。
核心实现仅需三步：
- 按主键分组：以 object_id 为分组依据；
- 指定透视字段：用 pivot("attr_name") 声明需展开的列名来源；
- 聚合取值：因同一 object_id + attr_name 可能存在多条记录（或空值），需指定聚合逻辑，推荐使用 F.first("value", ignorenulls=True) 或 F.first(F.col("value"))（Spark 3.4+ 默认忽略 null）。
✅ 完整代码示例：
from pyspark.sql import functions as F
# 假设 df 是原始 DataFrame（含 object_id, attr_name, value 等列）
result_df = (df
.groupBy("object_id")
.pivot("attr_name") # 自动提取所有唯一 attr_name 作为新列名
.agg(F.first("value")) # 取每组首个非空 value；若需严格去空，可加 ignorenulls=True
)
result_df.show(truncate=False)
⚠️ 注意事项：
- pivot() 会自动推断 attr_name 的全部唯一值作为列名，无需预先指定列列表，但要求该列基数适中（过高会导致大量稀疏列，影响性能与内存）；
- 若某些 object_id 缺失某属性（如 age 为空字符串或 null），对应单元格将显示 null，后续可用 fillna("") 统一处理；
- 在 Spark 3.0+ 中，pivot() 支持显式传入列名列表（如 .pivot("attr_name", ["name", "age", "number"])）以控制列顺序和范围，提升确定性；
- 性能优化建议：对 object_id 和 attr_name 字段建立适当分区（如 repartition("object_id")），避免 shuffle 倾斜。
总结：pivot() 是 Spark 中专为“长→宽”转换设计的声明式 API，语法简洁、语义清晰，配合 first() 聚合即可稳健处理含空值的真实业务数据，是构建特征宽表、报表中间层的首选方案。
