用 Scala Spark 比较两个值答案

【问题标题】：Compare two values with Scala Spark用 Scala Spark 比较两个值
【发布时间】：2021-04-26 18:35:01
【问题描述】：

我得到了下一个镶木地板文件：

+--------------+------------+-------+
|gf_cutoff     | country_id |gf_mlt |
+--------------+------------+-------+
|2020-12-14    |DZ          |5      |
|2020-08-06    |DZ          |4      |
|2020-07-03    |DZ          |4      |
|2020-12-14    |LT          |1      |
|2020-08-06    |LT          |1      |
|2020-07-03    |LT          |1      |

如您所见，由 country_id 参与，由 gf_cutoff DESC 排序。我想做的是比较 gf_mlt 以检查值是否已更改。为此，我想将最近的 gf_cutoff 与第二个进行比较。

这种情况的一个例子是比较：

 2020-12-14 DZ 5
with
 2020-08-06 DZ 4

我想在新列中写入，如果最近日期的值与第二行不同，则放入新列，DZ 的最新值为 5 并放入另一列 True if该值已更改，如果未更改，则为 false。进行此比较后，删除具有较旧行的行。

DZ 已经改变，而 LT 没有改变，因为都是 1。

所以输出会是这样的：

+--------------+------------+-------+------------+-----------+
|gf_cutoff     | country_id |gf_mlt | Has_change | old_value |
+--------------+------------+-------+------------+-----------+
|2020-12-14    |DZ          |5      |    True    |     4     |
|2020-12-14    |LT          |1      |    False   |     1     |

如果您需要更多解释，请告诉我。

【问题讨论】：

标签： scala dataframe apache-spark apache-spark-sql window-functions

【解决方案1】：

您可以在适当的窗口上使用lag 来获取最新值，然后在另一个适当的窗口上使用row_number 过滤最新的行：

import org.apache.spark.sql.expressions.Window

val df2 = df.withColumn(
    "last_value",
    lag("gf_mlt", 1).over(Window.partitionBy("country_id").orderBy("gf_cutoff"))
).withColumn(
    "rn", 
    row_number().over(Window.partitionBy("country_id").orderBy(desc("gf_cutoff")))
).filter("rn = 1").withColumn(
    "changed",
    $"gf_mlt" === $"last_value"
).drop("rn")

df2.show
+----------+----------+------+----------+-------+
| gf_cutoff|country_id|gf_mlt|last_value|changed|
+----------+----------+------+----------+-------+
|2020-12-14|        DZ|     5|         4|  false|
|2020-12-14|        LT|     1|         1|   true|
+----------+----------+------+----------+-------+

【讨论】：