Spark任务数优化咨询：60k Parquet文件下推过滤后任务数仍过高

先明确核心问题：Pushdown过滤只会剔除文件内的记录，但Spark默认会为每个输入文件分配一个任务（分区）——哪怕文件经过过滤后只剩极少数据。你调整的openCostInBytes是控制小文件合并的配置，但仅靠它不足以触发合并逻辑，得搭配其他手段或配置才行。

下面是具体解决办法：

1. 读取后直接合并分区 #

用coalesce或repartition在过滤操作后合并分区，快速减少任务数：

• coalesce是无shuffle合并，适合单纯减少分区数的场景，性能开销极小；
• repartition会触发shuffle，适合需要将数据均匀分配到新分区的情况。

示例代码（Scala）：

val df = spark.read.parquet("/your/data/path")
  .filter("your_filter_condition")
  .coalesce(1000) // 数值根据集群资源调整，比如每核保留1-2个任务

Spark优化器会自动将filter下推到读取阶段，所以先过滤再合并和先合并再过滤的性能差异不大，无需担心处理冗余数据。

2. 调整Spark小文件合并相关配置 #

除了openCostInBytes，还需搭配以下配置，让Spark自动合并小文件：

• spark.sql.files.maxPartitionBytes: 单个分区的最大字节数，默认128MB，可调至256MB或512MB，让Spark将多个小文件打包进一个分区；
• spark.sql.files.openCostInBytes: 默认4MB，代表打开一个文件的等效开销字节数，调至32MB左右，让Spark认为合并小文件的性价比更高；
• spark.sql.parquet.mergeSchema: 如果所有Parquet文件的schema一致，开启该配置（设为true），帮助Spark更好地识别可合并的文件组。

任务提交时的配置示例：

spark-submit \
  --conf spark.sql.files.maxPartitionBytes=268435456 \
  --conf spark.sql.files.openCostInBytes=33554432 \
  --conf spark.sql.parquet.mergeSchema=true \
  your_job.jar

3. 离线预合并小文件（根治方案） #

如果数据集是静态的或更新频率极低，直接在离线阶段合并小文件，从根源解决问题：

val df = spark.read.parquet("/original/data/path")
df.repartition(1000) // 根据总数据量/目标单分区大小估算，比如按128MB单分区计算
  .write
  .mode("overwrite")
  .parquet("/merged/data/path")

后续直接读取合并后的数据集，任务数会大幅下降。

4. 启用分区裁剪 #

如果你的Parquet数据是按目录分区的（比如按日期、地域），过滤时必须带上分区列条件，Spark会直接跳过不需要的分区目录，根本不会读取这些目录下的文件，任务数自然减少。比如数据按dt分区，过滤条件可写为filter("dt between '2024-01-01' and '2024-01-31'")，Spark只会扫描该时间范围内的文件。

5. 优化Parquet读取逻辑 #

• 确保spark.sql.parquet.enableVectorizedReader为true（默认开启），启用向量化读取，提升单任务处理效率，即便任务数未减少，整体运行时间也会缩短；
• 检查Parquet文件的统计信息：如果文件缺少正确的min/max等统计数据，Spark无法判断哪些文件可被过滤，只能全量读取。写入Parquet时可开启统计信息：

df.write
  .option("parquet.enable.summary-metadata", "true")
  .parquet("/your/data/path")

验证方式 #

调整后可查看Spark UI的Stage页面，确认输入分区数和任务数是否减少；再查看SQL页面的执行计划，验证filter是否已正确下推、文件合并逻辑是否生效。

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