AnalysisException: выражение не поддерживается в оконной функции при вычислении значения столбца на основе предыдущих строк

Question

У меня есть примеры данных, которые имеют 4 поля amt1, amt2, amt3 и amt4. Мы хотим вычислить значение amt5 на основе суммы полей (amt1, amt2, amt3, amt4) и значения amt5 предыдущей строки.

Предположим, ниже приведен набор данных:

+----+----+----+----+---+
|amt1|amt2|amt3|amt4|ids|
+----+----+----+----+---+
|   1|   2|   3|   4|  1|
|   1|   2|   3|   4|  2|
|   1|   2|   3|   4|  3|
|   1|   2|   3|   4|  4|
|   1|   2|   3|   4|  5|
|   1|   2|   3|   4|  6|
+----+----+----+----+---+

Ниже ожидаемый вывод:

+----+----+----+----+---+----+
|amt1|amt2|amt3|amt4|ids|amt5|
+----+----+----+----+---+----+
|   1|   2|   3|   4|  1|10  |
|   1|   2|   3|   4|  2|20  |
|   1|   2|   3|   4|  3|30  |
|   1|   2|   3|   4|  4|40  |
+----+----+----+----+---+----+

Ниже приведены исключения, которые я получаю после выполнения кода выше:

from pyspark.sql import Row
from pyspark.sql.window import Window
import pyspark.sql.functions as func

def sum(*col):
  sum = 0
  for i in col:
    sum = sum + i
  return sum

rdd = sc.parallelize(["1,1,2,3,4", "2,1,2,3,4", "3,1,2,3,4", "4,1,2,3,4", "5,1,2,3,4", "6,1,2,3,4"])
finalRdd = rdd.map(lambda t: t.split(",")).map(lambda t: Row(ids=t[0],amt1=t[1],amt2=t[2],amt3=t[3],amt4=t[4]))
df = spark.createDataFrame(finalRdd)

w = Window.orderBy("ids").rowsBetween(
    Window.unboundedPreceding,  # Take all rows from the beginning of frame
    Window.currentRow)          # To current row

df1 = df.withColumn("amt5",sum(df.amt1,df.amt2,df.amt3,df.amt4))
df1.withColumn("amt5",sum(df1.amt5).over(w)).show()

Ниже приведены исключения, которые я получаю после выполнения вышеуказанного кода:

py4j.protocol.Py4JJavaError: An error occurred while calling o121.withColumn.
: org.apache.spark.sql.AnalysisException: Expression '(amt5#11 + cast(0 as double))' not supported within a window function.;;
Project [amt1#0, amt2#1, amt3#2, amt4#3, ids#4, amt5#11, total#19]
+- Project [amt1#0, amt2#1, amt3#2, amt4#3, ids#4, amt5#11, total#19, total#19]
   +- Window [(amt5#11 + cast(0 as double)) windowspecdefinition(ids#4 ASC NULLS FIRST, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS total#19], [ids#4 ASC NULLS FIRST]
      +- Project [amt1#0, amt2#1, amt3#2, amt4#3, ids#4, amt5#11]
         +- Project [amt1#0, amt2#1, amt3#2, amt4#3, ids#4, ((((cast(amt1#0 as double) + cast(0 as double)) + cast(amt2#1 as double)) + cast(amt3#2 as double)) + cast(amt4#3 as double)) AS amt5#11]
            +- LogicalRDD [amt1#0, amt2#1, amt3#2, amt4#3, ids#4]

Answer 1

Вы столкнулись с коллизией в функции sum. Функция окна должна быть из пакета pyspark.sql.functions, поэтому вы должны вызывать ее следующим образом:

df1.withColumn("amt5",func.sum(df1.amt5).over(w)).show()