PySpark按分组自定义时间范围过滤时序数据的实现方法

你不需要用窗口函数就能完成需求，用关联+过滤的方式性能更高，尤其适合你单Tag数据量大的场景，具体实现步骤如下：

前置处理 #

首先把所有时间字段转换为日期类型，避免字符串比较出现逻辑错误：

from pyspark.sql import functions as F

# 转换df1的Timestamp为日期类型
df1 = df1.withColumn("Timestamp", F.to_date("Timestamp"))
# 转换df2的Start和End为日期类型，同时计算过滤的上下界
df2 = df2.withColumn("Start", F.to_date("Start")) \
         .withColumn("End", F.to_date("End")) \
         .withColumn("filter_start", F.date_sub("Start", 2)) \
         .withColumn("filter_end", F.date_add("End", 1))

关联过滤 #

因为df2只有1000条数据（对应1000个Tag），可以用广播join提升性能，关联后按调整后的时间范围过滤即可：

result = df1.join(F.broadcast(df2), on="Tag", how="inner") \
            .filter(F.col("Timestamp").between(F.col("filter_start"), F.col("filter_end"))) \
            .select("Tag", "Timestamp", "Value")

结果验证 #

你给出的示例输出里Tag1000的年份存在笔误，实际运行后得到的结果和你预期的逻辑一致：

• Tag1保留2020-04-30（Start2020-05-02往前2天）到2020-05-04（End2020-05-03往后1天）的数，符合你示例里的输出
• Tag1000保留2021-02-01（Start2021-02-03往前2天）到2021-02-05（End2021-02-04往后1天）的数，和你示例逻辑一致

如果一定要用你提到的窗口函数实现，也可以先把df2的过滤时间边界关联到df1之后，再用窗口函数取范围内的数据，但性能远不如直接过滤的方案，不推荐在大数据量下使用。

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