本文介绍一种不依赖 pivot 的高效方法，通过过滤、重命名与关联操作，将数据中某类特定行（如 col3='total'）的值提取为新列，适用于 spark sql 和 pyspark。
本文介绍一种不依赖 pivot 的高效方法，通过过滤、重命名与关联操作，将数据中某类特定行（如 col3='total'）的值提取为新列，适用于 spark sql 和 pyspark。
在实际数据处理中，我们常遇到“行转列”的需求，但标准的 pivot 操作通常要求对某一列进行分组聚合（如按 col1 分组，将 col3 的所有唯一值作为列名），而本例目标并非展开全部类别，而是仅将满足特定条件的一行（如 col3 == 'total'）的 col4 值，作为新列 total 关联到同组其他行上。此时 pivot 不仅冗余，还易因缺失值或分组逻辑偏差导致结果错误。
正确的思路是：分离 → 提取 → 关联。具体步骤如下：
-
分离数据：将原始 DataFrame 拆分为两部分
- 非总计行（col3 != 'total'）：保留原始结构，作为主表；
- 总计行（col3 == 'total'）：筛选后将 col4 重命名为 total，并丢弃原 col3（因已无意义）。
关联匹配：以业务主键（如 col1，也可扩展为 col1, col2 多列组合）为连接键，执行 left join（推荐）或 inner join。注意：若某些 col1 组没有对应的 'total' 行，使用 left join 可保留主表记录，total 列为 null（后续可用 coalesce 填充默认值，如 0）。
精简输出：显式选择主表全部字段 + 新增的 total 列，避免列名冲突或冗余字段。
以下是完整 PySpark 实现（支持 Spark 3.0+）：
from pyspark.sql import functions as F
# 假设 df 是原始 DataFrame
# 步骤1：提取非总计行（主表）
df_main = df.filter(F.col("col3") != "total")
# 步骤2：提取总计行，并构造 total 列
df_total = (df
.filter(F.col("col3") == "total")
.select("col1", "col2", F.col("col4").alias("total")))
# 步骤3：左连接（确保所有主表行都保留）
result = df_main.join(df_total, on=["col1", "col2"], how="left")
# 步骤4：处理缺失 total（可选：将 null 替换为 0）
result = result.withColumn("total", F.coalesce(F.col("total"), F.lit(0)))
# 查看结果
result.show()
✅ 关键说明：
- 若 col1 单独不足以唯一标识分组（如示例中 a,b 和 e,r 各有唯一 total），应使用 ["col1", "col2"] 多列连接，避免错误合并；
- 使用 left join 而非 inner join 更鲁棒——即使某组缺失 total 行，主数据也不会丢失；
- coalesce(..., lit(0)) 确保 total 列无空值，符合示例输出中 w,v 行 total=0 的要求；
- 此方案时间复杂度为 O(n)，远优于 pivot + groupBy 的聚合开销，且逻辑清晰、易于调试。
该方法本质是“查找映射表”，灵活通用：只需修改过滤条件（如 "col3" == "max" 或 "status" == "final"）和连接键，即可适配各类“单行标签转列”场景，无需编写 UDF 或复杂窗口函数。
