如何用PySpark将大数据集按日期聚合转换为基础指标数据集

我有如下格式的数据集：

希望转换为以下格式：

我现在写的代码是：

import pyspark
from pyspark.sql.functions import expr
#Create spark session
data = [("2023-01-01","10:30", 10, 20, 30), ("2023-01-01","11:30", 11, 21, 31) , \
      ("2023-01-01","13:30", 1, 2, 3),("2023-01-01","14:30", 110, 210, 310),("2023-01-02","01:30", 21, 21, 21), \
      ("2023-01-02","08:30", 11, 21, 31),("2023-01-02","11:30", 110, 210, 131),("2023-01-03","11:30", 10, 20, 30), \
      ("2023-01-03","12:30", 11, 21, 31),("2023-01-03","14:30", 8, 12, 13),("2023-01-03","15:30", 11, 21, 31)]

columns= ["Date","Time","Stock-a", "Stock-b", "Stock-c"]
df = spark.createDataFrame(data = data, schema = columns)
df.show()

from pyspark.sql.functions import expr, mean, stddev
columns = ["Stock-a", "Stock-b", "Stock-c"]
metrics_aggs = df.groupBy('Date').agg(
  *[mean(col).alias("mean_" + col) for col in columns],
  *[stddev(col).alias('std_' + col) for col in columns]
)
metrics_aggs.show()

现在需要找到一种列名透视的方法，最终只展示Mean和Stddev作为列，求解决思路或建议。

核心是把宽表转窄表（unpivot），再按日期+股票名称聚合计算指标，比先聚合再处理更简洁高效，步骤如下：

1. Unpivot宽表转长表：用stack函数把Stock-a、Stock-b、Stock-c拆成Stock Name（股票名）和Price（股价）两列，将多列数据合并为行级数据。
2. 分组聚合计算指标：按Date和Stock Name分组，直接计算均值和标准差，得到目标格式。

完整代码示例 #

import pyspark
from pyspark.sql.functions import expr, mean, stddev

# 创建Spark Session
spark = pyspark.sql.SparkSession.builder.appName("StockMetrics").getOrCreate()

data = [("2023-01-01","10:30", 10, 20, 30), ("2023-01-01","11:30", 11, 21, 31) , \
      ("2023-01-01","13:30", 1, 2, 3),("2023-01-01","14:30", 110, 210, 310),("2023-01-02","01:30", 21, 21, 21), \
      ("2023-01-02","08:30", 11, 21, 31),("2023-01-02","11:30", 110, 210, 131),("2023-01-03","11:30", 10, 20, 30), \
      ("2023-01-03","12:30", 11, 21, 31),("2023-01-03","14:30", 8, 12, 13),("2023-01-03","15:30", 11, 21, 31)]

columns= ["Date","Time","Stock-a", "Stock-b", "Stock-c"]
df = spark.createDataFrame(data = data, schema = columns)

# 1. 动态生成stack表达式（适配任意数量的Stock列）
stock_cols = [col for col in df.columns if col.startswith("Stock-")]
stack_expr = f"stack({len(stock_cols)}, {', '.join([f'{repr(col)}, {col}' for col in stock_cols])}) as (Stock_Name, Price)"
long_df = df.select("Date", expr(stack_expr)).dropna()

# 2. 分组计算均值和标准差
result_df = long_df.groupBy("Date", "Stock_Name").agg(
    mean("Price").alias("Mean"),
    stddev("Price").alias("Stddev")
)

# 展示结果
result_df.show()

补充说明 #

• 若需要总体标准差而非样本标准差，替换stddev为stddev_pop函数即可。
• 动态生成stack表达式的方式，无需手动修改代码即可适配新增的Stock列，扩展性更强。

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