比较 Spark 中当前行和上一行的值 [英] Compare Value of Current and Previous Row in Spark

问题描述

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

I am trying to compare record of current and previous row in the below DataFrame. I want to calculate the Amount column.

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).否则 AMOUNT of SL_NO - 1(即 - 50)
3. 对于第 3 行，如果 SL_NO - 3 和 2 的 ID 不相同，则需要考虑SL_NO 的数量 - 3(即 - 70).否则 AMOUNT of SL_NO - 2(即 - 30)

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

Same logic need to follow for the other rows also.

预期输出:

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

请帮忙.

推荐答案

你可以使用 lag 和 when.otherwise，这里是一个演示:

You could use lag with when.otherwise, here is a demonstration:

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是否排序在一起.

Note: since this example doesn't use any partition, it could have performance problem, in your real data, it would be helpful if your problem can be partitioned by some variables, may be Window.orderBy($"SL_NO").partitionBy($"ID") depending on your actual problem and whether IDs are sorted together.

这篇关于比较 Spark 中当前行和上一行的值的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！