PySpark中通过左连接条件填充null值的实现方案

核心需求 #

保留Series表所有行，为每行匹配对应series_id下后续步骤的Event值（例如series_id=a且step<4001的行，匹配后续step=4001对应的event=o）；或通过左连接后按规则填充event列的空值。

方案一：左连接后用窗口函数填充后续Event值 #

这种方法直接基于左连接结果，通过窗口函数提取后续第一个非空的Event值，逻辑简洁且高效。

步骤1：执行左连接 #

先将Series表与Events表按series_id和step做左外连接，保留Series表的所有行：

joined_df = series.join(events, ["series_id", "step"], "leftouter")

步骤2：用窗口函数填充空值 #

定义窗口规则：按series_id分区，按step升序排列，范围覆盖当前行到分区内最后一行。然后提取窗口内第一个非空的event值，填充当前行的空值：

from pyspark.sql import Window
from pyspark.sql.functions import col, first

# 定义窗口
window_spec = Window.partitionBy("series_id").orderBy("step").rowsBetween(0, Window.unboundedFollowing)

# 填充event列：取后续第一个非空的event值
filled_df = joined_df.withColumn(
    "event",
    first(col("event"), ignorenulls=True).over(window_spec)
)

步骤3：（可选）填充无匹配的默认值 #

如果某个series_id在Events表中无任何记录，event列会保留空值，可按需填充默认值：

filled_df = filled_df.fillna({"event": "unknown"})

步骤4：生成最终表 #

选择需要的字段输出：

final_df = filled_df.select("series_id", "step", "timestamp", "x", "y", "event")

方案二：预提取Event区间再关联 #

如果Events表中每个series_id的step是事件节点（例如step=4001是event=o的触发点），可以先为每个Event生成覆盖的step区间，再与Series表关联匹配。

步骤1：处理Events表生成区间 #

对Events表按series_id和step排序，用lead函数获取下一个Event的step（作为当前Event区间的上限）：

from pyspark.sql.functions import lead

events_with_range = events.withColumn(
    "next_step",
    lead(col("step")).over(Window.partitionBy("series_id").orderBy("step"))
)

# 最后一个Event的区间上限设为无穷大，覆盖所有后续step
events_with_range = events_with_range.fillna({"next_step": float("inf")})

步骤2：关联Series表与区间表 #

通过series_id关联，并匹配series.step落在当前Event的区间内（series.step < events.next_step）：

final_df = series.join(
    broadcast(events_with_range),
    (series.series_id == events_with_range.series_id) & (series.step < events_with_range.next_step),
    "leftouter"
).select(
    series.series_id,
    series.step,
    series.timestamp,
    series.x,
    series.y,
    events_with_range.event
).fillna({"event": "unknown"})

关键逻辑说明 #

• 方案一适用于需要匹配后续第一个出现的Event的场景，无需提前定义区间，灵活性更高。
• 方案二适用于Event对应明确step区间的场景，关联逻辑更直观。

内容的提问来源于stack exchange，提问作者Prabhjot Singh Rai