大量小Parquet文件拖慢Spark读取处理，求替代重写的标准方案

除了全量重写数据集，还有这些更高效的标准处理方式：

• 写入阶段提前控制文件大小
  从源头避免生成小文件：

  • 每日追加数据时，用coalesce或repartition将当日数据合并为合理大小的文件（建议128MB-256MB，匹配HDFS块大小），比如df.coalesce(1).write.mode("append").parquet(path)（根据数据量调整coalesce的数量）。
  • 配置Spark参数：spark.sql.files.maxRecordsPerFile设置单文件最大记录数；spark.sql.files.openCostInBytes调高文件打开成本阈值，让Spark在读取时更倾向于合并小文件的扫描任务。

• 分区+分桶表优化

  • 调整分区粒度：如果当前按日分区导致小文件过多，可改为按周/月分区，减少分区总数；同时在分区内使用分桶表（bucketBy），比如df.write.bucketBy(8, "user_id").sortBy("user_id").saveAsTable("table_name")，固定桶的数量，让数据均匀分布到桶中，避免每个分区生成大量小文件。
  • 分桶表还能让下游查询利用分桶 pruning，减少扫描的数据量。

• 增量合并已有小文件
  无需全量重写，只针对特定分区处理：

  • 定期对最近的分区（比如前一天的日期分区）执行合并，命令示例：

    val targetPath = "/path/to/partition=2024-05-20"
    spark.read.parquet(targetPath)
      .coalesce(1)
      .write.mode("overwrite")
      .parquet(targetPath)
    

  • 可以通过脚本遍历分区目录，只处理文件数量超过阈值的分区，降低资源消耗。

• 利用专用工具合并小文件

  • Hive环境下，直接执行ALTER TABLE table_name PARTITION (dt='2024-05-20') CONCATENATE;，该命令会后台合并同一分区内的Parquet小文件，无需全量重写。
  • 若使用Delta Lake，执行OPTIMIZE table_name WHERE dt='2024-05-20' ZORDER BY (event_time);，增量合并指定分区的小文件，同时按指定列排序优化查询性能，仅处理新增的小文件，效率更高。

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