Использование .where () в pyspark.sql.functions.max (). Over (window) в Spark 2.4 вызывает исключение Java

Question

Я последовал за публикацией StackOverflow о возврате максимума столбца, сгруппированного по другому столбцу, и получил неожиданное исключение 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$()))

Я пробовал это только на Spark 2.4 на Databricks.Я попробовал эквивалентный синтаксис SQL и получил ту же ошибку.

Answer 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|
+---+----+