PySpark优化：基于日期表过滤大表并按ID聚合求均值

问题背景 #

现有两个PySpark DataFrame：

• date_dataframe（月频）：包含字段from_date（当月首日）、to_date（from_date加1年）
• data_df（百万级日频）：包含字段id、p_date、value

需求目标：

1. 利用date_dataframe的时间窗口过滤data_df
2. 按id分组
3. 聚合计算value的均值

当前采用for循环实现，寻求更高效的无循环方案，允许调整date_dataframe格式

无循环实现方案 #

方案1：直接交叉关联+过滤（无需调整原date_dataframe） #

通过交叉连接关联两个表，再过滤出p_date落在对应时间窗口内的数据，最后完成分组聚合：

from pyspark.sql import functions as F

# 关联表并筛选符合时间窗口的记录
filtered_data = data_df.crossJoin(date_dataframe) \
    .filter(F.col("p_date").between(F.col("from_date"), F.col("to_date")))

# 按id分组计算均值
result = filtered_data.groupBy("id") \
    .agg(F.avg("value").alias("avg_value"))

方案2：调整date_dataframe为窗口数组格式（适合窗口数量较多的场景） #

将date_dataframe转换为包含所有时间窗口的数组结构，再通过数组展开完成关联过滤：

from pyspark.sql import functions as F

# 将所有时间窗口聚合为数组
window_array_df = date_dataframe.agg(
    F.collect_list(F.struct("from_date", "to_date")).alias("time_windows")
)

# 展开数组并关联数据，过滤符合条件的记录
filtered_data = data_df.crossJoin(window_array_df) \
    .select("id", "p_date", "value", F.explode("time_windows").alias("window")) \
    .filter(F.col("p_date").between(F.col("window.from_date"), F.col("window.to_date")))

# 分组计算均值
result = filtered_data.groupBy("id") \
    .agg(F.avg("value").alias("avg_value"))

性能优化提示 #

• 若date_dataframe存在重复或无效窗口，先做去重、过滤预处理，减少关联数据量
• 对data_df的p_date和id字段设置分区或建立索引，可大幅提升过滤和分组的效率
• 若时间窗口存在重叠，需确认需求：是保留所有符合窗口的记录后聚合，还是每条数据仅归属一个窗口（后者需额外处理窗口优先级逻辑）

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