如何为DataFrame添加含value列最大值的新列？

你遇到的问题是因为直接使用max($"value")这个聚合函数时，Spark需要你指定分组逻辑（比如groupBy），但你想要的是全局最大值并填充到每一行，所以得换两种更合适的思路来实现：

方法一：使用窗口函数（最常用的方案） #

窗口函数可以让你在不分组的情况下，为每行计算全局范围内的聚合值。步骤如下：

1. 先导入窗口函数相关的包：

import org.apache.spark.sql.expressions.Window

2. 定义一个覆盖全表的空窗口（不指定分区和排序，代表全局范围），再用max函数计算：

val globalWindow = Window.partitionBy() // 空分区表示全局范围
val df2 = df.withColumn("max_value", max($"value").over(globalWindow))

或者更简洁的写法，直接使用Window.all表示全局窗口：

val df2 = df.withColumn("max_value", max($"value").over(Window.all))

方法二：先计算全局最大值再关联 #

另一种思路是先算出全局最大值，再把这个值和原DataFrame做交叉关联：

1. 先计算全局最大值，得到一个仅包含一行一列的DataFrame：

val maxDF = df.select(max($"value").alias("max_value"))

2. 将原DataFrame和这个maxDF做交叉连接（因为maxDF只有一行，所以原表每行都会匹配到这个最大值），如果数据量较大，建议用broadcast优化性能：

import org.apache.spark.sql.functions.broadcast
val df2 = df.crossJoin(broadcast(maxDF))

为什么你的原代码无法运行？ #

原代码df.withColumn("max",max($"value"))会报错，因为Spark的聚合函数（比如max、sum）默认需要配合groupBy使用，用来对分组后的数据做聚合。如果没有指定分组，Spark无法确定聚合的范围，自然无法生成每行都包含的全局最大值列。

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