AWS EMR中Spark作业无法充分利用集群资源求助

这种情况我之前在EMR集群上也碰到过，大概率是并行度配置、资源参数或者代码写法没踩对路子，咱们一步步来排查解决：

1. 先优化代码写法 #

你把DataFrame转成RDD后用map操作其实没必要——DataFrame/Dataset有Spark Catalyst优化器加持，执行效率和资源利用率比原生RDD高很多。而且直接用RDD的map如果处理的是细粒度任务，很容易导致并行度不足，大部分节点闲下来。

建议改成DataFrame原生API写法：

val queries = spark.read.format("csv")
 .option("header", "true")
 .option("inferSchema", "true")
 .load("input_file.csv")

// 用DataFrame的map（如果doSomethingWith支持批量处理，换成mapPartitions效率更高）
val result = queries.map(q => doSomethingWith(q))

如果业务必须用RDD，那读完数据后一定要手动调整分区数，保证并行度匹配集群资源。

2. 调整数据分区，提升并行度 #

Spark作业的并行度由Task数决定，而Task数通常等于数据的分区数。如果你的CSV文件不大，默认分区数会很少，导致只有少数Task在跑，大部分集群资源闲置。

具体解决办法： #

• 读取时调整分区大小：设置spark.sql.files.maxPartitionBytes参数（默认128MB），比如改成64MB来增加分区数：

  spark.conf.set("spark.sql.files.maxPartitionBytes", "64mb")
  val queries = spark.read.format("csv")... // 后续读取逻辑
  

• 读取后手动重分区：根据集群总核数设置分区数（一般建议是总核数的2_{3倍），比如集群有32个可用核，就设64}96个分区：

  val queries = spark.read.format("csv")...load("input_file.csv").repartition(64)
  

3. 正确配置Spark提交的资源参数 #

在EMR上提交Spark作业时，必须明确指定executor的资源参数，不然YARN只会分配默认的少量资源（比如每个executor 1核1G内存），自然用不上集群的大部分资源。

比如提交命令可以这么写（根据你的集群实际资源调整）：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 8 \
  --executor-cores 4 \
  --executor-memory 16G \
  --driver-memory 4G \
  --conf spark.yarn.executor.memoryOverhead=2G \
  your-application.jar

参数说明：

• --num-executors：executor的数量，尽量用满集群从节点资源（主节点留少量资源给YARN管理）
• --executor-cores：每个executor的CPU核数，建议4~8核（避免过多上下文切换开销）
• --executor-memory：每个executor的内存，要给YARN留spark.yarn.executor.memoryOverhead（一般是executor内存的10%~20%）
• --driver-memory：驱动程序的内存，根据作业需求调整，不用太大

4. 检查EMR集群的YARN配置 #

有时候不是Spark的问题，是YARN本身的资源限制没放开：

• 登录EMR主节点，查看yarn-site.xml配置：
  • yarn.nodemanager.resource.cpu-vcores：要等于节点实际CPU核数（比如8核节点就设8）
  • yarn.nodemanager.resource.memory-mb：要等于节点实际内存（比如32G节点就设32768）
• 如果用CapacityScheduler，要确保Spark作业所在队列有足够资源配额（比如队列容量设为100%，或至少能容纳你配置的executor资源）

5. 用Spark UI精准排查 #

最后一定要打开Spark UI（EMR集群控制台能找到链接），看Stages页面：

• 查看每个Stage的Task数，如果远小于集群总核数，说明并行度不够，回到步骤2调整分区
• 看每个Task的运行时间，如果只有几秒，说明任务粒度太细，可以用mapPartitions代替map，批量处理数据减少调度开销

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