pyspark function.lag 条件答案

【问题标题】：pyspark function.lag on conditionpyspark function.lag 条件
【发布时间】：2021-03-12 22:21:57
【问题描述】：

我正在尝试解决 pyspark 的问题，

我有一个数据集，例如：

Condition | Date
0 | 2019/01/10
1 | 2019/01/11
0 | 2019/01/15
1 | 2019/01/16
1 | 2019/01/19
0 | 2019/01/23
0 | 2019/01/25
1 | 2019/01/29
1 | 2019/01/30

我想获取遇到condition == 1时日期列的最新滞后值

所需的输出类似于：

Condition | Date | Lag
0 | 2019/01/10 | NaN
1 | 2019/01/11 | NaN
0 | 2019/01/15 | 2019/01/11
1 | 2019/01/16 | 2019/01/11
1 | 2019/01/19 | 2019/01/16
0 | 2019/01/23 | 2019/01/19
0 | 2019/01/25 | 2019/01/19
1 | 2019/01/29 | 2019/01/19
1 | 2019/01/30 | 2019/01/29

我该怎么做？

请记住，它是一个非常大的数据集 - 我必须按 UUID 对其进行分区和分组，因此解决方案必须具有一定的性能。

谢谢，

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

这是 Pyspark 的解决方案。逻辑与@GordonLinoff 的 SQL 查询解决方案相同。

w = Window.orderBy("Date").rowsBetween(Window.unboundedPreceding, Window.currentRow - 1)
df.withColumn("Lag", max(when(col("Condition") == lit(1), col("Date"))).over(w)).show()

给予：

+---------+----------+----------+
|Condition|      Date|       Lag|
+---------+----------+----------+
|        0|2019/01/10|      null|
|        1|2019/01/11|      null|
|        0|2019/01/15|2019/01/11|
|        1|2019/01/16|2019/01/11|
|        1|2019/01/19|2019/01/16|
|        0|2019/01/23|2019/01/19|
|        0|2019/01/25|2019/01/19|
|        1|2019/01/29|2019/01/19|
|        1|2019/01/30|2019/01/29|
+---------+----------+----------+

【讨论】：

【解决方案2】：

在 SQL 中，可以使用条件运行max()：

select t.*,
       max(case when condition = 1 then date end) over (order by date
                                                        rows between unbounded preceding and 1 preceding
                                                       ) as prev_condition_1_date
from t;

【讨论】：

嗨！感谢您的快速响应，我正在尝试在 pyspark 上实现它，但我遇到了一些麻烦。

【解决方案3】：

我喜欢用 SQL 来解决这个问题：

from pyspark.sql.functions import expr

display(
  df.withColumn(
    'lag', 
    expr('max(case when Condition == 1 then Date end) over (order by Date rows between unbounded preceding and 1 preceding)'
  )
)

【讨论】：