如何使用PySpark按最接近时间戳合并两个DataFrame？

问题根源 #

直接做交叉连接会生成笛卡尔积（比如200条×200条=40000条记录），这就是你看到大量冗余数据的原因。必须先计算每条记录与另一表中时间戳的差值，再筛选出每个记录对应的最接近匹配。

实现步骤 #

1. 统一时间戳类型 #

先确保两个DataFrame的time列是PySpark的TimestampType，否则没法计算时间差：

from pyspark.sql import functions as F
from pyspark.sql.types import TimestampType

# 转换time列类型
df1 = df1.withColumn("time", F.col("time").cast(TimestampType()))
df2 = df2.withColumn("time", F.col("time").cast(TimestampType()))

# 重命名避免列名冲突（比如把df2的time和ID改成time2、ID2）
df2 = df2.withColumnRenamed("time", "time2").withColumnRenamed("ID", "ID2")

2. 给每条记录加唯一标识 #

为了后续精准筛选每个记录的最匹配项，给两个表的记录分别加唯一ID：

df1 = df1.withColumn("record_id_1", F.monotonically_increasing_id())
df2 = df2.withColumn("record_id_2", F.monotonically_increasing_id())

3. 交叉连接并计算时间差 #

将两个表交叉连接，然后计算两条记录时间戳的绝对差值（用秒数计算更直观）：

cross_df = df1.crossJoin(df2)

# 计算时间差绝对值（秒）
cross_df = cross_df.withColumn(
    "time_diff",
    F.abs(F.unix_timestamp("time") - F.unix_timestamp("time2"))
)

4. 筛选每个记录的最接近匹配 #

用窗口函数给每个record_id_1的匹配项按时间差排序，取排名第一的（即时间差最小的）：

from pyspark.sql.window import Window

# 按record_id_1分组，按time_diff升序排序
window_spec = Window.partitionBy("record_id_1").orderBy("time_diff")

# 给每个分组内的记录排名，取排名第一的
result_df = cross_df.withColumn("rank", F.row_number().over(window_spec)) \
                    .filter(F.col("rank") == 1) \
                    .drop("record_id_1", "record_id_2", "time_diff", "rank")

5. 同ID内匹配的优化（如果需要） #

如果你的实际需求是相同ID下找最接近的时间戳，不要用交叉连接，改成按ID内连接：

# 按ID连接，只匹配同ID的记录
cross_df = df1.join(df2, on="ID", how="inner")
# 后续计算时间差、窗口筛选的步骤和上面一致

性能优化建议 #

如果数据量较大，直接交叉连接可能卡慢，可以先加时间范围过滤，减少匹配数量：

# 只匹配时间差在1小时内的记录（可根据需求调整）
cross_df = df1.crossJoin(df2).filter(
    F.abs(F.unix_timestamp("time") - F.unix_timestamp("time2")) <= 3600
)

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