Spark分组聚合：按ID统计Threshold之上的Value总和

原始数据表 #

目标结果表 #

我已经想到一种临时解决方案：新增标记列筛选出大于Threshold的Value，再对符合条件的值进行聚合。但Spark是否提供无需新增列的更优实现方式（比如窗口函数）？

解决方案：无需新增列的聚合实现 #

当然有更简洁高效的方式，不用额外加标记列，直接在聚合逻辑里完成条件判断即可，比窗口函数更适合这个场景（窗口函数会保留所有行，后续还要去重，反而冗余）。

方式1：sum结合when条件求和 #

Spark的聚合函数支持直接嵌套条件判断，一行代码就能完成分组求和：

// Scala 版本
import org.apache.spark.sql.functions.{sum, when}

df.groupBy("ID", "Threshold")
  .agg(
    sum(when($"Value" > $"Threshold", $"Value").otherwise(0)).alias("total_above_threshold")
  )
  .show()

# PySpark 版本
from pyspark.sql.functions import sum, when

df.groupBy("ID", "Threshold")
  .agg(
    sum(when(df.Value > df.Threshold, df.Value).otherwise(0)).alias("total_above_threshold")
  )
  .show()

方式2：filter配合分组聚合（可读性拉满） #

也可以直接对分组内的Value列做过滤后求和，写法更直观：

// Scala 版本
df.groupBy("ID", "Threshold")
  .agg(
    sum($"Value".filter($"Value" > $"Threshold")).alias("total_above_threshold")
  )
  .show()

# PySpark 版本
from pyspark.sql.functions import sum

df.groupBy("ID", "Threshold")
  .agg(
    sum(df.Value.filter(df.Value > df.Threshold)).alias("total_above_threshold")
  )
  .show()

关于窗口函数的说明 #

如果一定要用窗口函数实现，虽然可行但完全没必要——因为窗口函数会给每一行都附上聚合结果，最后还要手动去重，多了额外步骤，性能不如直接分组聚合。示例如下（仅作演示，不推荐）：

// Scala 窗口函数写法（冗余，不推荐）
import org.apache.spark.sql.functions.{sum, when}
import org.apache.spark.sql.expressions.Window

val windowSpec = Window.partitionBy("ID", "Threshold")

df.withColumn("total_above_threshold", sum(when($"Value" > $"Threshold", $"Value").otherwise(0)).over(windowSpec))
  .select("ID", "Threshold", "total_above_threshold")
  .distinct()
  .show()

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