如何解决EMR上Spark加载大量小文件导致任务卡顿的问题？

这个问题我太熟悉了——大量小文件绝对是Spark任务的性能杀手！你遇到的情况完全符合小文件带来的典型问题：第一次任务处理190个大文件时，分区数合理，Task调度和I/O都高效；但输出的小文件让第二次任务不得不启动成百上千个Task，每个Task光初始化和打开文件的时间就远超过计算时间，直接拖垮了整个集群。下面给你几个实用的解决方案，分补救、预防和优化三个维度：

一、先补救：合并已有的小文件 #

如果已经生成了大量小文件，先跑一个轻量任务把它们合并成大文件，再进行后续处理：

• 用Spark的coalesce合并（无shuffle，更高效）：coalesce可以在不触发shuffle的情况下减少分区数，适合把小文件合并成指定数量的大文件。比如你初始是190个文件，我们可以合并成相同数量的文件（每个约260MB）：

// Scala示例，Python同理
val smallFilesDF = spark.read.text("hdfs://path/to/your/small-files-output")
smallFilesDF.coalesce(190)
  .write
  .mode("overwrite")
  .text("hdfs://path/to/merged-large-files")

如果数据分布不均匀，也可以用repartition（会触发shuffle，但分区更均衡），但优先选coalesce。

• 用Hadoop DistCp工具合并：如果文件在HDFS上，也可以用DistCp这个专门的文件复制/合并工具，通过设置map数来控制合并后的文件数量：

hadoop distcp -Dmapreduce.job.maps=190 \
  hdfs://source/small-files-path hdfs://target/merged-files-path

二、从根源预防：优化第一次任务的输出 #

解决小文件最好的方式是不要让它生成，调整第一次任务的输出配置：

• 手动指定输出分区数：根据总数据量和目标文件大小计算分区数（比如每个文件256MB-1GB），50GB的话设为190-200个分区就很合适：

// 第一次任务写数据时添加repartition
yourProcessedDF.repartition(190)
  .write
  .mode("overwrite")
  .text("hdfs://path/to/optimized-output")

• 启用Spark内置的小文件合并配置：在Spark提交任务时添加这些参数，让Spark自动处理小文件：

  • --conf spark.sql.shuffle.partitions=190：调整shuffle后的分区数，默认200，和你的初始文件数匹配
  • --conf spark.sql.files.maxRecordsPerFile=1000000：限制每个输出文件的最大记录数，避免生成过小的文件
  • --conf spark.sql.files.openCostInBytes=10485760：调大文件打开成本的阈值（默认4MB），让Spark更倾向于合并小文件

• 用分桶表存储（适合后续有查询需求的场景）：如果后续任务经常按某个字段过滤/聚合，用bucketBy分桶既能避免小文件，还能提升后续查询效率：

yourProcessedDF.write
  .mode("overwrite")
  .bucketBy(190, "your-key-column")
  .saveAsTable("your_bucketed_table")

三、应急优化：处理小文件时的配置调整 #

如果不得不直接处理小文件，调整以下Spark配置来缓解性能问题：

• 调大分区文件大小阈值：--conf spark.sql.files.maxPartitionBytes=268435456（默认128MB，改为256MB），让Spark把多个小文件合并成一个分区，减少Task数量
• 避免过多小Task：--conf spark.executor.cores=4和--conf spark.executor.instances=10（根据集群规模调整），让每个Executor处理更多Task，减少调度开销

这些方案的核心逻辑都是减少Task数量、降低I/O开销——毕竟Spark的优势是处理大数据，单个Task处理的数据量越大，性能优势越明显。

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