Spark 2.4x在AWS EMR自动扩缩容时如何优化default.parallelism与sql.shuffle

首先明确：Spark的spark.default.parallelism和spark.sql.shuffle.partitions这类核心参数，启动时的配置会被Spark UI环境页快照记录，但运行时修改全局配置不会同步到UI显示，也不会让已启动的Executor自动应用新值——你之前在Driver里设置没看到UI变化是正常的，但不代表实际执行没生效，得通过任务执行计划或Stage分区数验证。

下面分场景给出可行方案：

一、应用内动态调整的可行操作 #

1. 针对SQL/Shuffle操作调整spark.sql.shuffle.partitions #

你可以在Driver的Spark会话中动态设置会话级参数，后续的SQL查询、Shuffle类操作（比如join、groupBy）会自动使用新值：

// Scala示例：根据当前Executor数量计算合适的shuffle分区数
val executorCount = spark.sparkContext.getExecutorMemoryStatus.size - 1 // 减去Driver本身
val corePerExecutor = 4 // 替换为你的EMR Executor核数
val newShufflePartitions = executorCount * corePerExecutor * 2 // 经验值，可根据业务调整
spark.conf.set("spark.sql.shuffle.partitions", newShufflePartitions.toString)

注意：Spark UI环境页不会更新这个值，但你可以查看SQL执行计划（Explain）里的Shuffle节点分区数，或者Stage页面的Shuffle Read/Write分区数，确认新参数已生效。

2. 针对RDD操作调整分区数 #

spark.default.parallelism是RDD的默认分区数，运行时修改全局配置不会生效，但可以：

• 创建RDD时显式指定分区数，比如sc.parallelize(data, newPartitionNum)
• 对已有RDD重分区，用rdd.repartition(newPartitionNum)（会触发Shuffle）或rdd.coalesce(newPartitionNum)（不触发Shuffle，仅减少分区）

二、联动EMR自动扩缩容的实现 #

要在EMR自动扩缩容事件触发时自动调整参数，你可以：

• 监听扩缩容事件：通过CloudWatch Events捕获EMR集群的实例增减事件，触发Lambda函数，Lambda向Driver发送信号（比如通过Redis、ZooKeeper或者Driver暴露的HTTP接口）
• Driver主动轮询：在Driver中定期调用spark.sparkContext.getExecutorMemoryStatus获取当前活跃Executor数量，自动计算并调整分区参数
• 两种方式的核心都是：根据当前集群资源（Executor数、核数）动态计算合适的分区数，然后在会话内设置参数或对RDD重分区

三、关于Master节点设置的误区 #

EMR的Master节点是集群管理节点，不负责单个Spark应用的运行时参数调整，所有应用级的配置都由Driver管理，所以不需要去Master节点做任何设置——你之前的推测是错误的。

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