Spark处理大量小文件性能优化：EMR批量转Gzip到Parquet

针对你遇到的驱动端文件索引和任务拆分耗时过长的问题，以下是几个实际可行的优化方案：

1. 启用Spark并行文件元数据发现 #

Spark默认由驱动单线程遍历文件系统收集元数据，当文件数量达到百万级时会成为瓶颈。开启并行元数据发现后，会将文件路径拆分给执行器并行收集，大幅缩短耗时：

• 设置参数：

  spark.sql.sources.parallelPartitionDiscovery.threshold=10000
  spark.sql.sources.parallelPartitionDiscovery.parallelism=200
  

  • threshold：当文件数超过该值时触发并行发现（建议设为10000以上）
  • parallelism：并行度，建议设置为集群总执行器核心数的1.5-2倍

2. 预先合并小Gzip文件 #

Gzip是不可拆分的压缩格式，每个文件对应一个Spark任务，百万级文件会导致任务调度开销剧增。可以先合并小文件再处理：

• 使用EMR的DistCp并行合并：

  hadoop distcp -Dmapreduce.job.maps=100 s3://your-source-bucket/small-gz/ s3://your-temp-bucket/merged-gz/
  

  • -Dmapreduce.job.maps：设置并行合并的Map任务数，根据集群资源调整
• 合并后再处理大文件，既能减少元数据收集量，也能降低任务调度开销

3. 优化S3元数据访问（若使用S3存储） #

如果文件存储在S3，通过以下配置减少S3 API调用的开销：

• 开启S3Guard元数据缓存：

  spark.hadoop.fs.s3a.metastore.impl=org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore
  spark.hadoop.fs.s3a.s3guard.ddb.table.name=your-s3guard-table
  

  利用DynamoDB缓存文件元数据，避免每次遍历都调用S3 List API
• 调整S3列表并行度：

  spark.hadoop.fs.s3a.list.status.parallelism=100
  

  提升S3客户端的并行列表能力

4. 明确指定分区路径减少遍历范围 #

如果文件按分区目录存储（如按日期分层），直接指定分区路径而非递归遍历根目录：

• 示例：

  spark.read.text("s3://bucket/path/year=202*/month=*")
  

  让Spark直接读取指定分区的元数据，避免遍历所有子目录

5. 升级Spark/EMR版本 #

Spark 3.x及以上版本对文件元数据收集和任务拆分做了大量优化，EMR 6.x系列默认搭载Spark 3.x。如果当前使用的是EMR 5.x（对应Spark 2.x），升级后能显著提升该阶段的性能

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