Scala实现Spark DataFrame按measure_type值pivot并应用不同聚合操作

实现方案 #

你可以通过两种常用方式实现该需求，运行结果完全匹配你给出的预期输出： #

方式1：Spark Scala DSL 实现

推荐显式指定pivot的枚举值，避免Spark额外扫描全表统计维度取值，提升执行性能：

import org.apache.spark.sql.functions._

// 1. 定义已知的measure_type枚举值
val measureTypes = Seq("measureA", "measureB", "measureC")

val result = df.groupBy("ts", "key")
  // 2. 按measure_type pivot转列
  .pivot("measure_type", measureTypes)
  // 3. 对每个维度应用对应聚合规则
  .agg(
    sum("value").alias("sum"),
    avg("value").alias("avg"),
    max("value").alias("max")
  )
  // 4. 提取需要的列，重命名为目标格式
  .select(
    col("ts"),
    col("key"),
    col("measureA_sum").alias("measureA"),
    col("measureB_avg").alias("measureB"),
    col("measureC_max").alias("measureC")
  )

// 打印结果验证
result.show()

方式2：Spark SQL 实现

逻辑更直观易懂：

// 注册临时视图
df.createOrReplaceTempView("measure_data")

// 执行SQL查询
val result = spark.sql("""
SELECT
  ts,
  key,
  SUM(CASE WHEN measure_type = 'measureA' THEN value END) AS measureA,
  AVG(CASE WHEN measure_type = 'measureB' THEN value END) AS measureB,
  MAX(CASE WHEN measure_type = 'measureC' THEN value END) AS measureC
FROM measure_data
GROUP BY ts, key
ORDER BY ts, key
""")

// 打印结果验证
result.show()

两种方式执行后输出的结果均为：

+---+---+--------+--------+--------+
| ts|key|measureA|measureB|measureC|
+---+---+--------+--------+--------+
|  1| k1|       6|     6.0|       7|
|  2| k1|    null|     8.0|       9|
|  2| k2|       5|    null|       8|
+---+---+--------+--------+--------+

如果需要统一数值类型，可以额外调用cast函数对avg结果做类型转换即可。

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