如何让EMR集群所有Executor读取.csv及.csv.gz文件？

• Gzip压缩的单线程限制：.csv.gz属于单线程压缩格式，Spark无法拆分单个gzip文件进行并行读取，只能启动一个Task处理整个文件，自然只会用到一个Executor。就算是未压缩的.csv，如果是单个超大文件，Spark默认按文件大小和spark.sql.files.maxPartitionBytes计算的初始分区数可能远小于你的Executor数量，导致并行度不足。
• 初始分区数不足：Spark读取文件时，默认根据总文件大小除以spark.sql.files.maxPartitionBytes（默认128MB）生成初始分区。如果这个计算结果远小于8个Executor的总处理能力，就没法让所有Executor参与读取。

处理Gzip压缩文件的核心方案 #

• 拆分大gzip文件：在上传到存储路径前，把单个大gzip文件拆成多个小文件（每个大小接近128MB）。这样Spark会为每个小文件分配一个独立Task，多个Executor就能并行读取。
• 替换为可拆分的压缩格式：如果后续数据生成环节可以调整，换成Snappy、LZO这类支持拆分的压缩格式，Spark能直接拆分单个大压缩文件生成多个Task，无需提前拆分文件。

调整Spark配置提升并行度 #

• 强制指定读取后的分区数：读取CSV后立刻用repartition设置足够的分区数，建议设为总Executor核数的1-2倍（m5.2xlarge是8核，8台总核数64，可设为64或128）。示例代码：

val rawDF = spark.read.option("header", "true").csv("s3://your-input-path/")
val parallelDF = rawDF.repartition(64)

• 调整分区字节数配置（仅对未压缩CSV有效）：如果是未压缩的超大CSV，减小spark.sql.files.maxPartitionBytes的值，让Spark生成更多初始分区。比如改成32MB：

spark.conf.set("spark.sql.files.maxPartitionBytes", "32m")

配套资源配置检查 #

确认EMR的Spark资源配置：m5.2xlarge有8vCPU，建议每个Executor分配4核+对应内存，这样8台实例可以提供16个Executor，配合足够的Task数就能把所有资源用起来。

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