PySpark中基于ID与时间戳差的条件关联实现问题

首先要明确，Spark中处理这类带范围条件的连接，更高效的方式是先按id做等值连接，再过滤时间差条件。你的代码问题主要在于两个点：没有对时间差取绝对值，以及直接在join的on参数中混用非等值条件（部分Spark版本对这种写法支持有限，且性能不佳）。

步骤1：确保时间字段为Timestamp类型 #

如果你的timestamp1和timestamp2是字符串格式，先转换为Spark的Timestamp类型：

from pyspark.sql import functions as F

df1 = df1.withColumn("timestamp1", F.to_timestamp("timestamp1"))
df2 = df2.withColumn("timestamp2", F.to_timestamp("timestamp2"))

步骤2：执行连接并过滤时间差 #

先按id等值连接，再过滤两个时间戳的绝对差值小于10分钟（600秒）的记录：

df_joined = df1.join(df2, on="id", how="inner") \
    .filter(F.abs(df1.timestamp1.cast("long") - df2.timestamp2.cast("long")) < 600)

或者用unix_timestamp函数实现同样的效果：

df_joined = df1.join(df2, on="id", how="inner") \
    .filter(F.abs(F.unix_timestamp("timestamp1") - F.unix_timestamp("timestamp2")) < 600)

结果验证 #

执行上述代码后，得到的df_joined会和你期望的输出一致：

id, timestamp1, timestamp2
a, 2023-01-01 10:00:00, 2023-01-01 10:05:00

原代码问题说明 #

1. 缺少绝对值处理：原代码只判断了timestamp1比timestamp2早10分钟内的情况，忽略了timestamp2更早的场景，加上abs()才能覆盖双向的时间差判断。
2. 非等值条件放在join的on中：Spark对on参数中的非等值连接支持有限，先做等值join再过滤，不仅兼容性更好，执行效率也更高。

内容的提问来源于stack exchange，提问作者Movilla