PySpark-SQL循环优化：月度日期过滤场景替代逐天循环的提速方案

核心优化逻辑是避免循环中重复扫描原表、重复提交Spark任务的开销，以下是可落地的实现方案：

方案1：一次性区间匹配+单趟写入（提升幅度最大，推荐优先使用） #

原逻辑每次循环都要全表扫描一次，整月30次循环相当于扫描原表30次，该方案仅需扫描1次原表即可完成全月计算：

1. 先生成整月目标日期的小体量序列表
2. 前置过滤原表中可能符合全月条件的候选数据，减少参与计算的数据量
3. 通过区间条件一次性关联出所有日期匹配的记录，直接按分区写入即可

代码示例：

from pyspark.sql import functions as f

# 构造整月目标日期序列，仅30条数据，体量极小
target_dates = list(range(1, 31))
date_df = spark.createDataFrame([(d,) for d in target_dates], schema=["load_date"])
# 广播小表避免shuffle开销
date_df = f.broadcast(date_df)

# 前置谓词过滤，只保留可能命中整月任意一天的记录
source_df = spark.sql("select * from table where x_date <= 30 and y_date >= 1")

# 一次性匹配所有符合区间条件的记录
result_df = source_df.join(
    date_df,
    on=(source_df.x_date <= date_df.load_date) & (source_df.y_date >= date_df.load_date),
    how="inner"
)

# 开启动态分区一次性写入所有分区
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
# 开启自适应执行自动合并小文件，减少写入开销
spark.conf.set("spark.sql.adaptive.enabled", "true")

result_df.write.partitionBy("load_date").mode("overwrite").parquet(output_path)

其他辅助优化点 #

• 如果原表是分区存储表，可以在前置过滤阶段加上分区裁剪条件，进一步减少扫描的数据量
• 调整spark.sql.shuffle.partitions参数适配数据量级，避免任务并行度不足导致的长尾
• 若存在单条记录覆盖大量日期导致的数据倾斜，可以对原表数据加盐打散后再关联，解决倾斜问题

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