scala - 如何添加具有最大值的新列？

Question

我有一个包含 2 列 tag 和 value 的 Dataframe。

我想添加一个新列，其中包含 value 列的 max。 (每一行的值都相同)。

我尝试过如下操作，但没有成功。

val df2 = df.withColumn("max",max($"value"))

如何将 max 列添加到数据集？

Answer 1

有 3 种方法可以做到(您已经从另一个答案中知道了一种)。我避免使用 collect，因为它并不是真正需要的。

这里是最大值3出现两次的数据集。

val tags = Seq(
  ("tg1", 1), ("tg2", 2), ("tg1", 3), ("tg4", 4), ("tg3", 3)
).toDF("tag", "value")
scala> tags.show
+---+-----+
|tag|value|
+---+-----+
|tg1|    1|
|tg2|    2|
|tg1|    3| <-- maximum value
|tg4|    4|
|tg3|    3| <-- another maximum value
+---+-----+

笛卡尔加入“最大”数据集

我将使用 tags 的笛卡尔连接和具有最大值的单行数据集。

val maxDF = tags.select(max("value") as "max")
scala> maxDF.show
+---+
|max|
+---+
|  4|
+---+
val solution = tags.crossJoin(maxDF)
scala> solution.show
+---+-----+---+
|tag|value|max|
+---+-----+---+
|tg1|    1|  4|
|tg2|    2|  4|
|tg1|    3|  4|
|tg4|    4|  4|
|tg3|    3|  4|
+---+-----+---+

我不担心这里的笛卡尔连接，因为它只是一个单行数据集。

窗口聚合

我最喜欢的窗口聚合非常适合这个问题。另一方面，我真的不认为这是最有效的方法，因为使用的分区数量很多，即只有 1 个，这会导致最差的并行性。

诀窍是在空窗口规范上使用聚合函数 max，通知 Spark SQL 以任何顺序使用所有行。

val solution = tags.withColumn("max", max("value") over ())
scala> solution.show
18/05/31 21:59:40 WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
+---+-----+---+
|tag|value|max|
+---+-----+---+
|tg1|    1|  4|
|tg2|    2|  4|
|tg1|    3|  4|
|tg4|    4|  4|
|tg3|    3|  4|
+---+-----+---+

请注意说明一切的警告。

WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.

考虑到其他解决方案，我不会使用这种方法，并将其留在这里用于教育目的。