火花滞后与默认值作为另一列

Posted 2023-03-31

【中文标题】火花滞后与默认值作为另一列

【英文标题】：Spark lag with default value as another column

【发布时间】：2019-04-05 14:22:38

【问题描述】：

假设我们有

|bin | min |  end  | start | 
|1   | 5   | 10    | 
|2   | 12  | 24    | 
|3   | 28  | 36    | 
|4   | 40  | 50    | 
|5   | null| null  | 

我想将 start 填充为上一列的 end 以生成连续的 bin 值。对于缺少的我想用当前的 min 来填写。对于空行，我考虑单独处理。

延迟会给我们带来什么

df.withColumn("start", F.lag(col("end"), 1, ***default_value***).over(orderBy(col("bin"))
|bin | min |  end  | start | 
|1   | 5   | 10    | (5 wanted)
|2   | 12  | 24    | 10
|3   | 28  | 36    | 24
|4   | 40  | 50    | 36
|5   | null| null  | null

我的问题：

1/ 我们在 default_value 中放入什么来延迟获取当前行的另一列，在本例中为 min

2/ 有没有办法同时处理空行而不分开？我打算过滤 non-null ，执行滞后，然后与空行联合。如果 Null 是第一个（bin 1）或最后一个（bin 5），答案会有什么不同？

【参考方案1】：

使用coalesce 获取组中第一行的列值。

from pyspark.sql import functions as F 
df.withColumn("start", F.coalesce(F.lag(col("end"), 1).over(orderBy(col("bin")),col("min")))

lag 当前不支持ignorenulls 选项，因此您可能需要分离出null 行，计算非空行的start 列和数据帧的union。