比较 Spark 中当前行和上一行的值

Question

我正在尝试比较以下DataFrame中当前和上一行的记录。我想计算金额列。

scala> val dataset = sc.parallelize(Seq((1, 123, 50), (2, 456, 30), (3, 456, 70), (4, 789, 80))).toDF("SL_NO","ID","AMOUNT")
scala> dataset.show
+-----+---+------+
|SL_NO| ID|AMOUNT|
+-----+---+------+
|    1|123|    50|
|    2|456|    30|
|    3|456|    70|
|    4|789|    80|
+-----+---+------+

计算逻辑：

1. 对于第 1 行，AMOUNT 应为第一行的 50。
2. 对于第 2 行，如果 SL_NO - 2 和 1 的 ID 不同，则需要考虑 SL_NO - 2(即 - 30)的量。否则 SL_NO - 1 的金额(即 - 50)
3. 对于第 3 行，如果 SL_NO - 3 和 2 的 ID 不同，则需要考虑 SL_NO量 - 3(即 - 70)。否则金额 SL_NO - 2 (即 - 30)

其他行也需要遵循相同的逻辑。

预期输出：

+-----+---+------+
|SL_NO| ID|AMOUNT|
+-----+---+------+
|    1|123|    50|
|    2|456|    30|
|    3|456|    30|
|    4|789|    80|
+-----+---+------+

请帮忙。

Answer 1

您可以将lag与when.otherwise一起使用，这是一个演示：

import org.apache.spark.sql.expressions.Window
val w = Window.orderBy($"SL_NO")
dataset.withColumn("AMOUNT", 
when($"ID" === lag($"ID", 1).over(w), lag($"AMOUNT", 1).over(w)).otherwise($"AMOUNT")
).show
+-----+---+------+
|SL_NO| ID|AMOUNT|
+-----+---+------+
|    1|123|    50|
|    2|456|    30|
|    3|456|    30|
|    4|789|    80|
+-----+---+------+

注意：由于此示例不使用任何分区，因此可能存在性能问题，在您的实际数据中，如果您的问题可以通过某些变量进行分区，可能会Window.orderBy($"SL_NO").partitionBy($"ID")，具体取决于您的实际问题以及ID是否排序在一起。