PySpark多条件关联DataFrame并分组取最近上报对应值的实现方法

错误原因 #

你之前的写法报错是因为在executor端执行的rdd.map算子内部调用了需要driver端调度的DataFrame操作（filter、collect），属于Spark不支持的嵌套作业调用，会触发锁或者序列化异常，该写法需完全规避。

前置准备 #

先确保两个DataFrame的date字段均为DateType类型，而非字符串，避免日期比较逻辑异常。

实现方案 #

方案1：窗口函数实现（兼容所有PySpark版本）

逻辑和你提供的原生SQL逻辑一致，性能稳定：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 条件左连接：同GROUP且B的日期小于等于A的日期
join_df = A.join(
    B,
    (A["GROUP"] == B["GROUP"]) & (B["date"] <= A["date"]),
    how="left"
)

# 定义窗口：按A的GROUP、A的date分区，B的date降序排序
win = Window.partitionBy(A["GROUP"], A["date"]).orderBy(B["date"].desc())

# 取每个窗口排名第一的记录（即B的最大日期对应的记录）
result = join_df.withColumn("rn", F.row_number().over(win)) \
    .filter(F.col("rn") == 1) \
    .select(
        A["GROUP"],
        A["date"],
        B["date"].alias("last_reported_val"),
        B["val"]
    ).orderBy("GROUP", "date")

方案2：ASOF JOIN实现（PySpark 3.0+ 推荐）

专门针对时间序列最近值匹配场景设计，性能远高于先关联再开窗的方案，适合大数据量场景：

# ASOF JOIN要求两个表先按分组字段和时间字段排序
A_sorted = A.sort("GROUP", "date")
B_sorted = B.sort("GROUP", "date")

# 执行ASOF JOIN，自动匹配同GROUP下小于等于A.date的最近B表记录
result = A_sorted.join(
    B_sorted,
    on=["GROUP"],
    how="left",
    on_asof="date",
    asof_by="GROUP"
).select(
    "GROUP",
    A["date"],
    B["date"].alias("last_reported_val"),
    "val"
)

两种方案输出结果均和你提供的预期结果完全一致。

内容的提问来源于stack exchange，提问作者Mμ.