继承满足条件的最近行的值答案

【问题标题】：Carrying over values from most recent row satisfying a condition继承满足条件的最近行的值
【发布时间】：2020-12-03 03:28:53
【问题描述】：

我有一个 pyspark 数据框，它看起来像（一个更大的版本）如下：

+---+---+----+----+
| id|  t|type| val|
+---+---+----+----+
|100|  1|   1|  10|
|100|  2|   0|NULL|
|100|  5|   1|  20|
|100|  8|   0|NULL|
|100| 12|   0|NULL|
|100| 20|   0|NULL|
|100| 22|   1|  30|
|200|  5|   1|  40|
|200| 11|   0|NULL|
|200| 19|   1|  50|
|200| 24|   0|NULL|
|200| 25|   0|NULL|
+---+---+----+----+

我想创建一个新列，对于 1 类型的行，使用 val，对于 0 类型，使用来自 1 类型的最新条目的 val。

输出如下所示：

+---+---+----+----+----+
| id|  t|type| val|val2|
+---+---+----+----+----+
|100|  1|   1|  10|  10|
|100|  2|   0|NULL|  10|
|100|  5|   1|  20|  20|
|100|  8|   0|NULL|  20|
|100| 12|   0|NULL|  20|
|100| 20|   0|NULL|  20|
|100| 22|   1|  30|  30|
|200|  5|   1|  40|  40|
|200| 11|   0|NULL|  40|
|200| 19|   1|  50|  50|
|200| 24|   0|NULL|  50|
|200| 25|   0|NULL|  50|
+---+---+----+----+----+

如果我们在 pandas 数据框中，如何通过迭代来完成这一点相当简单，但我无法找到使用 pyspark 工具的方法。我想做的是

from pyspark.sql import Window
import pyspark.sql.functions as sf

w = Window.partitionBy(['id']).orderBy('t')
df.withColumn('val2',
  sf.when(col('type'), col('val')).otherwise(sf.lag(col('val')).over(w))
)

但这会产生

+---+---+----+----+----+
| id|  t|type| val|val2|
+---+---+----+----+----+
|100|  1|   1|  10|  10|
|100|  2|   0|NULL|  10|
|100|  5|   1|  20|  20|
|100|  8|   0|NULL|  20|
|100| 12|   0|NULL|NULL|
|100| 20|   0|NULL|NULL|
|100| 22|   1|  30|  30|
|200|  5|   1|  40|  40|
|200| 11|   0|NULL|  40|
|200| 19|   1|  50|  50|
|200| 24|   0|NULL|  50|
|200| 25|   0|NULL|NULL|
+---+---+----+----+----+

我明白为什么这不起作用，但我不知道如何解决它。我想我可以使用 groupby('id').applyInPandas(...) 和一个遍历行的函数，但这会很慢，有没有更好的方法？

【问题讨论】：

我觉得你可以先用最后一个值 (SO post) 填充空值，然后用多个 when 语句 (link) 实现你的 val2 col

标签： python pyspark

【解决方案1】：

是的，您可以将最后一个函数与 ignorenulls=True 一起使用：

from pyspark.sql import Window
import pyspark.sql.functions as sf

w = Window.partitionBy(['id']).orderBy('t')
df.withColumn('val2',
sf.when(col('type'), 
col('val')).otherwise(sf.last(col('val'),ignorenulls=True).over(w)))

【讨论】：