Glue ETL Spark作业并行度远超预期的原因咨询

你的假设错误，问题出在你使用了Scala标准库的并行集合（par.map），而非Spark的分布式并行执行逻辑。

Scala的par.map是基于Driver端JVM的多线程并行机制，它的并行度由JVM的线程池（默认是ForkJoinPool）控制，和你配置的Spark Executor核心数完全无关。你设置的spark.executor.cores=4是控制分布式Task在Executor上的并行数，但dfList.par.map的逻辑是在Driver进程内启动多线程来逐个处理DataFrame，这些任务属于Driver端的本地线程任务，不是Spark提交到Executor的分布式Task——这就是为什么Spark UI显示10个并行任务的原因。

• RDD/DataFrame的分区数：这是决定并行任务数的核心，每个分区对应一个分布式Task。比如spark.default.parallelism=4是shuffle操作后的默认分区数，但如果源DataFrame的分区数更多，实际并行任务数会以源分区数为准。
• 集群总可用核心数：即spark.executor.instances × spark.executor.cores，这是集群能同时运行的最大分布式Task数（每个Task占用1核）。你当前的配置是1个Executor×4核，所以最多同时运行4个分布式Task，但总Task数可以远大于这个数（Task会排队执行）。
• Shuffle操作的分区配置：对于join、groupBy这类shuffle操作，并行数由spark.sql.shuffle.partitions（默认值200）控制，该配置会覆盖spark.default.parallelism。
• Spark任务调度机制：Spark的TaskScheduler会在有Task完成后立即调度新的Task补位，所以总Task数可以远超同时运行的Task数，但并行运行的峰值不会超过集群总核心数。
• 动态资源分配：如果开启了动态资源分配（spark.dynamicAllocation.enabled=true），Spark会根据任务需求自动增减Executor数量，进而调整总核心数和并行处理能力。

如果想让DataFrame的处理利用Spark的分布式并行能力，不要用Scala的par.map，可以选择以下方式：

• 将多个DataFrame合并为一个（比如unionByName），然后用Spark的分布式操作统一处理；
• 使用Spark的foreachPartition或mapPartitions操作，基于分区进行并行处理；
• 通过spark.submitJob API异步提交多个DataFrame的处理任务，让Spark调度到Executor上分布式执行。

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