PySpark中如何实现多列动态透视表？

当然可以实现动态多列透视！你之前的尝试失败是因为sf.mean()无法直接接收列名列表——它仅支持单个Column对象作为参数。下面给你两种灵活的动态实现方案，都能生成你期望的输出格式：

方案一：宽表转长表再透视（更直观可控） #

这种方法先把需要透视的多列转为键值对的长表格式，再通过透视组合出目标列名，逻辑清晰且列名调整更灵活：

import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql import functions as sf

# 初始化Spark会话（如果还没创建的话）
spark = SparkSession.builder.appName("MultiColumnPivot").getOrCreate()

# 你的原始DataFrame
sdf = spark.createDataFrame(
    pd.DataFrame([[1,2,6,1],[1,3,3,2],[1,6,0,3],[2,1,0,1],
                  [2,1,7,2],[2,7,8,3]], columns = ['id','val1','val2','month'])
)

# 定义需要透视的列（完全动态，可随意添加/修改）
col_to_pivot = ['val1', 'val2']

# 1. 将宽表转为长表：用stack函数把val1、val2转为metric（列名）和value（对应值）
stack_expr = sf.stack(len(col_to_pivot), 
                      *[item for pair in zip(col_to_pivot, col_to_pivot) for item in pair]) \
              .alias("metric", "value")
sdf_long = sdf.select('id', 'month', stack_expr)

# 2. 同时透视metric和month，聚合取唯一值（因为每个id+metric+month组合只有一条数据）
sdf_pivot = sdf_long.groupBy('id') \
                    .pivot(['metric', 'month']) \
                    .agg(sf.first('value'))

# 3. 重命名列：把(val1,1)格式转为val1_month1
sdf_pivot = sdf_pivot.select(
    'id',
    *[sf.col(c).alias(f"{c[0]}_month{c[1]}") for c in sdf_pivot.columns[1:]]
)

# 查看结果
sdf_pivot.show()

执行后就能得到你想要的格式：

+---+-------------+-------------+-------------+-------------+-------------+-------------+
| id|val1_month1  |val1_month2  |val1_month3  |val2_month1  |val2_month2  |val2_month3  |
+---+-------------+-------------+-------------+-------------+-------------+-------------+
|  1|            2|            3|            6|            6|            3|            0|
|  2|            1|            1|            7|            0|            7|            8|
+---+-------------+-------------+-------------+-------------+-------------+-------------+

方案二：动态生成聚合表达式（无需转表） #

如果你不想转换表结构，可以动态生成每个列的聚合表达式，再通过调整列名得到目标格式：

# 定义需要透视的列
col_to_pivot = ['val1', 'val2']

# 动态生成聚合表达式：为每个列创建mean（或first）表达式并指定别名
agg_exprs = [sf.mean(c).alias(c) for c in col_to_pivot]

# 执行透视：pivot后列名会自动格式化为「月份_列名」
sdf_pivot = sdf.groupBy('id') \
               .pivot('month') \
               .agg(*agg_exprs)

# 重命名列：把「1_val1」改为「val1_month1」
new_columns = ['id'] + [f"{c.split('_')[1]}_month{c.split('_')[0]}" for c in sdf_pivot.columns[1:]]
sdf_pivot = sdf_pivot.toDF(*new_columns)

# 查看结果
sdf_pivot.show()

两种方案的对比 #

• 方案一：逻辑直观，列名调整更灵活，适合需要透视的列较多或后续可能扩展的场景；
• 方案二：无需转换表结构，性能略优，但列名处理依赖PySpark的默认命名规则，需要注意格式匹配。

无论哪种方案，只要修改col_to_pivot列表就能轻松添加或移除需要透视的列，完全满足动态需求。

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