Spark在EMR节点启动超配Executor的问题咨询

我发现Spark在单个节点上启动的Executor数量经常超出预期，这导致节点负载平均值飙升，最终引发executor lost、bad node或unhealthy node问题。

我的Spark配置与环境 #

• 每个节点配置8核、32GB内存；master、core及task节点的yarn.nodemanager.resource.cpu-vcores设为6，单个节点最多应运行6个Executor。
• 部分作业使用2核8GB配置，部分使用1核2GB配置——理论上最坏情况下单个节点最多启动6个容器，但上图显示实际启动了9个。
• 通过Ganglia观察到节点负载平均值达19（节点仅8个vcore），高负载导致Executor丢失（Driver无法按时接收心跳；Executor因高负载运行缓慢）。
• 使用环境为EMR 6.9、Spark 3.2.0。

问题 #

1. 为何Spark启动的Executor数量超出配置限制？
2. 能否在此场景下将单个节点的Executor最大数量限制为6？
3. 另一节点有5核、25GB内存可用，为何Spark/YARN未选择在该节点启动Executor？

提交给AWS EMR的启动配置JSON文件 #

master、core及task节点使用相同配置：

我设置较高的心跳等参数是为避免Executor丢失，作业运行时长为1-1.5小时，若Executor能维持运行则作业可完成，否则之前的努力将白费。

[  
  {    
    "Classification": "yarn-site",    
    "Properties": {      
      "yarn.nodemanager.resource.cpu-vcores": "6",      
      "yarn.nodemanager.resource.memory-mb": "30000",      
      "yarn.resourcemanager.nodemanagers.heartbeat-interval-max-ms":"60000",      
      "yarn.resourcemanager.nodemanagers.heartbeat-interval-min-ms":"60000",      
      "yarn.resourcemanager.nodemanagers.heartbeat-interval-ms":"60000",      
      "yarn.nodemanager.health-checker.timeout-ms":"72000000",      
      "yarn.nodemanager.health-checker.interval-ms":"36000000",      
      "yarn.resourcemanager.application-timeouts.monitor.interval-ms":"180000"    
    }  
  },  
  {    
    "Classification": "mapred-site",    
    "Properties": {      
      "yarn.app.mapreduce.am.scheduler.heartbeat.interval-ms":"60000",      
      "yarn.app.mapreduce.am.hard-kill-timeout-ms":"600000"    
    }  
  },  
  {    
    "Classification": "spark-defaults",    
    "Properties": {      
      "spark.executor.heartbeatInterval": "600s",      
      "spark.network.timeout":"7200s"    
    }  
  }
]

问题解答 #

1. 为何Executor数量超出配置限制？

核心原因是YARN与Spark的资源调度逻辑存在缺口，加上配置遗漏：

• yarn.nodemanager.resource.cpu-vcores是YARN对外暴露的可调度核数，但当节点仍有物理核剩余时，YARN可能不会严格拦截超出该值的调度请求，尤其是你的Executor采用1核小规格配置时，总核数会“挤”满物理8核，绕过6核的配置限制。
• EMR节点上运行的系统守护进程（如Ganglia、YARN服务）会占用部分资源，但Spark/YARN调度时未将这些预留资源计算在内，导致过度分配。

2. 能否限制单个节点的Executor最大数量为6？

完全可以，两种实现方式：

• 作业级别配置：提交作业时添加--conf spark.yarn.max.executor.per.node=6，直接限制单节点Executor数量上限，不受单个Executor核数影响。
• 集群级别配置：在spark-defaults分类中添加spark.yarn.max.executor.per.node=6，所有作业默认遵循该限制，无需单次作业单独配置。

3. 为何有空闲资源的节点未被选中启动Executor？

主要由YARN调度策略和状态同步延迟导致：

• 本地性偏好：Spark默认优先将Executor调度到有数据本地性的节点（如数据存储在该节点HDFS），即使该节点负载高，YARN也会优先调度，直到本地性等待超时。可调整spark.locality.wait参数（默认3s）缩短等待时间，让YARN更快切换到空闲节点。
• 心跳间隔过长：你设置的60s心跳间隔会导致ResourceManager无法及时感知空闲节点的资源状态，调度决策基于旧数据，不会向空闲节点分配Executor。建议适当降低心跳间隔（如10s），平衡开销与状态同步及时性。
• 队列/标签限制：若作业绑定了特定YARN队列或节点标签，而空闲节点不在该范围，YARN也不会调度到该节点，可检查EMR的队列和节点标签配置。

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