在 Spark 2.4 上的 pyspark.sql.functions.max().over(window) 上使用 .where() 会引发 Java 异常

Posted 2023-04-15

【中文标题】在 Spark 2.4 上的 pyspark.sql.functions.max().over(window) 上使用 .where() 会引发 Java 异常

【英文标题】：Using .where() on pyspark.sql.functions.max().over(window) on Spark 2.4 throws Java exception

【发布时间】：2019-02-03 23:24:44

【问题描述】：

我在*** 上关注了关于返回由另一列分组的列的最大值的帖子，但遇到了意外的 Java 异常。

这是测试数据：

import pyspark.sql.functions as f
data = [('a', 5), ('a', 8), ('a', 7), ('b', 1), ('b', 3)]
df = spark.createDataFrame(data, ["A", "B"])
df.show()

+---+---+
|  A|  B|
+---+---+
|  a|  5|
|  a|  8|
|  a|  7|
|  b|  1|
|  b|  3|
+---+---+

这是据称适用于其他用户的解决方案：

from pyspark.sql import Window
w = Window.partitionBy('A')
df.withColumn('maxB', f.max('B').over(w))\
    .where(f.col('B') == f.col('maxB'))\
    .drop('maxB').show()

应该产生这个输出：

#+---+---+
#|  A|  B|
#+---+---+
#|  a|  8|
#|  b|  3|
#+---+---+

相反，我得到：

java.lang.UnsupportedOperationException: Cannot evaluate expression: max(input[2, bigint, false]) windowspecdefinition(input[0, string, true], specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$()))

我只在 Databricks 上的 Spark 2.4 上试过这个。我尝试了等效的 SQL 语法并得到了同样的错误。

【参考方案1】：

Databricks 支持能够在 Spark 2.4 上重现该问题，但不能在早期版本上重现。显然，这是由于制定物理计划的方式不同（如果需要，我可以发布他们的回复）。计划进行修复。

同时，这里是原始问题的另一种解决方案，它不属于 2.4 版本的问题：

df.withColumn("maxB", f.max('B').over(w)).drop('B').distinct().show()

+---+----+
|  A|maxB|
+---+----+
|  b|   3|
|  a|   8|
+---+----+

【讨论】：

顺便说一句，如果您在withColumn 之后persist 您的数据框，则不会发生错误。