Spark SQL/PySpark中用Pivot替代Group By+CASE实现行转列

需求回顾 #

原始长表数据需要将Component的不同取值（70、260）转为单独列，列值对应Tabel字段，最终得到宽表结构。你已经通过CASE WHEN + GROUP BY实现需求，但希望用更简洁的Pivot方案优化，避免冗余的聚合函数。

Spark SQL 实现方案 #

Spark SQL原生支持PIVOT语法，刚好适配你的场景。由于复合主键（OmgevingID、AdministratieKantoorID、WerkgeverID、JaarID、VolgnummerRun、PersoneelsnummerVerloning、Periode + Component）保证每个分组下单个Component仅对应一条Tabel记录，用FIRST()聚合函数更贴合语义（替代冗余的MAX()）：

SELECT *
FROM YourTableName
PIVOT (
    FIRST(Tabel)
    FOR Component IN (70 AS Component_70_tabel, 260 AS Component_260_tabel)
)
ORDER BY PersoneelsnummerVerloning;

说明：

• PIVOT子句中，FIRST(Tabel)指定取唯一对应的Tabel值；
• FOR Component IN (...)明确要转为列的Component值，并定义新列名；
• 无需手动写GROUP BY，Spark会自动根据除Component外的所有列分组。

PySpark DataFrame API 实现方案 #

如果用PySpark代码实现，逻辑类似，先指定分组列，再执行透视操作，最后重命名列：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder.appName("component_pivot").getOrCreate()

# 读取原始表数据
df = spark.read.table("YourTableName")

# 执行透视：分组列 + 透视Component列 + 聚合取Tabel值
pivoted_df = df.groupBy(
    "OmgevingID", "AdministratieKantoorID", "WerkgeverID",
    "AdministratieID", "JaarID", "VolgnummerRun",
    "PersoneelsnummerVerloning", "Periode", "Datum"
).pivot("Component", [70, 260]).agg({"Tabel": "first"})

# 重命名列以匹配需求格式
pivoted_df = pivoted_df.withColumnRenamed("70", "Component_70_tabel") \
                      .withColumnRenamed("260", "Component_260_tabel")

# 查看结果
pivoted_df.show()

说明：

• groupBy()指定所有非透视列作为分组键；
• pivot("Component", [70,260])指定透视字段和目标取值列表；
• agg({"Tabel": "first"})聚合逻辑和Spark SQL一致，取唯一对应的Tabel值。

方案优势 #

相比CASE WHEN + GROUP BY，Pivot方案：

• 语法更直观，将透视逻辑集中在一处，代码更简洁；
• 当需要扩展更多Component值时，只需修改IN子句或透视取值列表，无需新增多个CASE WHEN分支；
• 用FIRST()替代MAX()，更符合数据唯一性的语义，避免冗余感。

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