Scala实现Spark DataFrame按name分组各列类别取最大分钟值优化

Great question! Your current approach gets the job done, but we can simplify it significantly by removing redundant operations and streamlining the logic—while producing exactly the result you need.

问题分析 #

First, let's break down what your code is actually doing (and where it can be optimized):

• Your window functions + max(sumnt) combo is redundant: when you use sum(min).over(Window.partitionBy("name","nw")), every row in the same name+nw group gets the same sum value. Taking the max() of these identical values doesn't change anything—you can directly compute the sum with a simple groupBy instead.
• The multiple joins are necessary to get all combinations of nw, cat, and pt per name, but we can clean up the preceding aggregation steps to make the code more efficient.

优化后的代码 #

Here's a streamlined version that achieves your goal in fewer steps:

val customers = sc.parallelize(Seq(
  ("Alice", "abc","cat1","pt1", 50.00), 
  ("Alice", "abc","cat1","pt1", 45.00), 
  ("Alice", "bcd","cat2","pt1", 55.00), 
  ("Bob", "abc","cat1","pt4", 25.00), 
  ("Bob", "bcd","cat1","pt4", 29.00), 
  ("Bob", "av","cat4","pt4",27.00)
)).toDF("name","nw","cat","pt","min")

// Step 1: 计算每个(name, nw)、(name, cat)、(name, pt)分组的分钟总和
val nwAgg = customers.groupBy("name", "nw").agg(sum("min").alias("sum_nw"))
val catAgg = customers.groupBy("name", "cat").agg(sum("min").alias("sum_cat"))
val ptAgg = customers.groupBy("name", "pt").agg(sum("min").alias("sum_pt"))

// Step 2: 按name关联三个聚合结果，得到每个name下nw/cat/pt的所有组合
val result = nwAgg
  .join(catAgg, Seq("name"), "inner")
  .join(ptAgg, Seq("name"), "inner")

// 查看结果
result.show()

结果验证 #

这段代码会生成和你原方案完全一致的最终输出：

+-----+---+-------+----+-------+---+-------+
| name| nw|sum_nw | cat|sum_cat| pt|sum_pt |
+-----+---+-------+----+-------+---+-------+
|Alice|bcd| 55.0  |cat1| 95.0  |pt1| 150.0 |
|Alice|bcd| 55.0  |cat2| 55.0  |pt1| 150.0 |
|Alice|abc| 95.0  |cat1| 95.0  |pt1| 150.0 |
|Alice|abc| 95.0  |cat2| 55.0  |pt1| 150.0 |
| Bob| av| 27.0  |cat1| 54.0  |pt4| 81.0  |
| Bob| av| 27.0  |cat4| 27.0  |pt4| 81.0  |
| Bob|bcd| 29.0  |cat1| 54.0  |pt4| 81.0  |
| Bob|bcd| 29.0  |cat4| 27.0  |pt4| 81.0  |
| Bob|abc| 25.0  |cat1| 54.0  |pt4| 81.0  |
| Bob|abc| 25.0  |cat4| 27.0  |pt4| 81.0  |
+-----+---+-------+----+-------+---+-------+

为什么这更优？ #

• 减少冗余操作: 移除了无意义的窗口函数和max()计算，直接用分组求和完成核心逻辑。
• 逻辑更清晰: 代码直接对应你的需求——先计算各维度的总和，再组合所有可能的维度组合。
• 性能更高效: 更少的中间数据处理步骤，在大数据量场景下执行速度会更快。

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