Apache Spark动态分配未达预期问题排查咨询（AWS EMR环境）

问题背景 #

在AWS EMR集群运行Spark作业，配置中启用了动态分配，设置spark.dynamicAllocation.maxExecutors=10、spark.dynamicAllocation.minExecutors=2，但集群无其他作业运行时，仅分配2个Executor，且存在大量闲置处理器资源。作业配置如下：

spark.app.id    ABC
spark.app.name  XYZ
spark.app.startTime 1699623174356
spark.blacklist.decommissioning.enabled true
spark.blacklist.decommissioning.timeout 1h
spark.decommissioning.timeout.threshold 20
spark.default.parallelism   100
spark.driver.cores  4
spark.driver.defaultJavaOptions -XX:OnOutOfMemoryError='kill -9 %p'
spark.driver.extraClassPath ****
spark.driver.extraJavaOptions   *********(redacted)
spark.driver.extraLibraryPath   *****
spark.driver.host   ****
spark.driver.memory 20g
spark.driver.port   41783
spark.driver.userClassPathFirst false
spark.dummy.for.ops 548510a92b094285a4d3568ada11c44a
spark.dynamicAllocation.enabled true
spark.dynamicAllocation.maxExecutors    10
spark.dynamicAllocation.minExecutors    2
spark.emr.default.executor.cores    4
spark.emr.default.executor.memory   18971M
spark.eventLog.dir  *****
spark.eventLog.enabled  true
spark.executor.cores    4
spark.executor.defaultJavaOptions   -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'
spark.executor.extraClassPath   /lib/spark/jars/*:/home/hadoop/extrajars/*:/var/log/log4j2-config.xml
spark.executor.extraJavaOptions *********(redacted)
spark.executor.extraLibraryPath *********
spark.executor.id   driver
spark.executor.memory   20g
spark.executor.memoryOverhead   1500
spark.executor.userClassPathFirst   false
spark.executorEnv.AAA_APPLICATION_NAME  DoomSpark
spark.executorEnv.AAA_AWS_CLIENT_RELATIONSHIP_KEY_BUCKET_NAME   aaastack-prod-na-syncbucketcfnbucketc68d5f32-1mafjvqn9i9gj
spark.executorEnv.CORAL_CONFIG_PATH /home/hadoop/.config/coral-config
spark.executorEnv.DOMAIN    prod
spark.executorEnv.ENVROOT   /home/hadoop
spark.executorEnv.REALM USAmazon
spark.executorEnv.REDIS_AUTH_TOKEN  *********(redacted)
spark.executorEnv.REDIS_PORT    6379
spark.executorEnv.REDIS_URL master.rediscfnreplicationgroup-prod-na.f12qkp.use1.cache.amazonaws.com
spark.files.fetchFailure.unRegisterOutputOnHost true
spark.hadoop.fs.s3.getObject.initialSocketTimeoutMilliseconds   2000
spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version.emr_internal_use_only.EmrFileSystem    2
spark.hadoop.mapreduce.fileoutputcommitter.cleanup-failures.ignored.emr_internal_use_only.EmrFileSystem true
spark.hadoop.mapreduce.output.fs.optimized.committer.enabled    true
spark.hadoop.yarn.timeline-service.enabled  false
spark.history.fs.logDirectory   hdfs:///var/log/spark/apps
spark.history.ui.port   18080
spark.master    yarn
spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY_HOSTS  ip-10-0-52-16.ec2.internal
spark.org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter.param.PROXY_URI_BASES  http://ip-10-0-52-16.ec2.internal:20888/proxy/application_1699353854257_0781
spark.resourceManager.cleanupExpiredHost    true
spark.scheduler.mode    FAIR
spark.serializer    org.apache.spark.serializer.KryoSerializer
spark.shuffle.service.enabled   true
spark.sql.emr.internal.extensions   com.amazonaws.emr.spark.EmrSparkSessionExtensions
spark.sql.hive.metastore.sharedPrefixes com.amazonaws.services.dynamodbv2
spark.sql.parquet.fs.optimized.committer.optimization-enabled   true
spark.sql.parquet.output.committer.class    com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter
spark.sql.warehouse.dir hdfs:///user/spark/warehouse
spark.stage.attempt.ignoreOnDecommissionFetchFailure    true
spark.submit.deployMode cluster
spark.submit.pyFiles    
spark.ui.filters    org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter
spark.ui.port   0
spark.yarn.app.container.log.dir    /var/log/hadoop-yarn/containers/application_1699353854257_0781/container_1699353854257_0781_01_000001
spark.yarn.app.id   application_1699353854257_0781
spark.yarn.appMasterEnv.DOMAIN  prod
spark.yarn.appMasterEnv.REALM   USAmazon
spark.yarn.appMasterEnv.SPARK_PUBLIC_DNS    $(hostname -f)
spark.yarn.executor.memoryOverheadFactor    0.1875
spark.yarn.heterogeneousExecutors.enabled   false
spark.yarn.historyServer.address    ip-XXXXX.internal:XXXX
spark.yarn.submit.waitAppCompletion false
spark.yarn.tags livy-batch-780-2iPUo7av

可能的原因 #

• 作业并行度不足：Spark动态分配的核心逻辑是根据待处理任务数扩容。当前配置每个Executor有4个core，2个Executor可提供8个并行任务槽。如果作业的实际任务数（如RDD分区数、SQL shuffle分区数）≤8，现有Executor足以处理所有任务，不会触发扩容。
• 动态分配触发条件未满足：Spark需要持续的任务积压才会扩容。默认需等待spark.dynamicAllocation.schedulerBacklogTimeout（1秒）的pending任务，之后每间隔spark.dynamicAllocation.sustainedSchedulerBacklogTimeout（5秒）请求新增Executor。若作业任务执行过快，无持续积压，不会触发扩容。
• YARN队列资源限制：即使集群有闲置资源，作业所在YARN队列的capacity或maximum-capacity参数可能限制了资源配额。比如队列仅允许分配8个vcore，刚好对应2个Executor，无法扩容到10个。
• 集群可用资源不足：每个Executor需消耗4vcore + 21.5g内存（20g executor内存 + 1.5g overhead）。EMR会为系统服务预留部分资源（每个节点默认1vcore + 10%内存），若集群总可用资源不足以支撑10个Executor，YARN无法分配更多。
• 作业阶段特性限制：若作业是单阶段任务（如简单读写无shuffle），或大部分任务在少数阶段快速完成，没有持续的任务压力，Spark不会启动更多Executor。

排查步骤 #

• 检查作业任务并行度：
  • 进入Spark UI的Stages页面，查看各阶段的任务总数、并行执行数、pending任务时长。若任务数少或执行快，说明无需更多Executor。
  • 对于SQL作业，检查spark.sql.shuffle.partitions（默认200）是否适配数据量；对于RDD作业，确认分区数是否足够（建议设为集群总core数的2-3倍）。
• 验证YARN资源分配情况：
  • 在EMR主节点执行yarn top，查看集群总闲置vcore和内存。
  • 执行yarn queue -status default（默认队列），查看队列的资源配额和已使用情况，确认是否有配额限制。
  • 查看YARN ApplicationMaster日志（EMR日志目录或CloudWatch），搜索是否有资源申请被拒绝的日志。
• 查看动态分配触发日志：
  • 查看Spark Driver日志，搜索dynamic allocation、requesting executors、executor added关键字，确认是否触发扩容请求，或未触发的原因（如No pending tasks, not requesting executors）。
• 计算集群可用资源：
  • 统计所有Core/Task节点的总资源，减去系统预留资源，计算可容纳的Executor数量。例如m5.2xlarge节点（8vcore/32g）预留后可用7vcore/28.8g，仅能容纳1个当前配置的Executor，若集群仅2个此类节点，最多分配2个Executor。
• 调整并行度测试：
  • 手动设置更高并行度，如spark.default.parallelism=40（10个Executor×4core）或spark.sql.shuffle.partitions=40，重新提交作业，观察Executor数量是否增加。
• 确认动态分配参数：
  • 在Spark UI的Environment页面，查看spark.dynamicAllocation.schedulerBacklogTimeout、spark.dynamicAllocation.sustainedSchedulerBacklogTimeout等参数的实际值，确认是否因参数设置导致扩容条件过严。

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