☰

企业官网建设流程全解析

首页 / 建站日记 / 企业官网建设流程全解析
Spark SQL Shuffle 分区数生成机制详解
2026/7/21 9:05:32 网站建设 项目流程

前言

在 Spark SQL 物理计划已经生成之后、真正执行之前，还会通过一系列 rule（如 AQE、EnsureRequirements、WSCG 等），对整棵物理计划进行全局整合或优化。

普通 Join、Group By 等算子所需的隐式 Shuffle，通常是在EnsureRequirements中补充的。

例如，一个聚合 SQL 生成的初始物理计划是：

HashAggregateExec(final) +- HashAggregateExec(partial) +- FileSourceScanExec

经过EnsureRequirements后：

HashAggregateExec(final) +- ShuffleExchangeExec(HashPartitioning(user_id, 200)) +- HashAggregateExec(partial) +- FileSourceScanExec

Spark SQL 的 Shuffle 分区数，通常是在执行EnsureRequirements并插入ShuffleExchangeExec时确定的。

接下来，本文基于 Spark 3.1.2 源码视角，详细剖析 Spark SQL 发生 Shuffle 时的分区数决定机制。

EnsureRequirements介绍

org.apache.spark.sql.execution.exchange.EnsureRequirements

objectEnsureRequirementsextendsRule[SparkPlan]{privatedefensureDistributionAndOrdering(operator:SparkPlan):SparkPlan={// 1、获取当前算子要求 child 需要满足的数据分布和排序要求valrequiredChildDistributions:Seq[Distribution]=operator.requiredChildDistributionvalrequiredChildOrderings:Seq[Seq[SortOrder]]=operator.requiredChildOrderingvarchildren:Seq[SparkPlan]=operator.children assert(requiredChildDistributions.length==children.length)assert(requiredChildOrderings.length==children.length)// 2、判断当前算子 child 的数据分布是否满足当前算子的输入要求，并根据不同结果作处理// Ensure that the operator's children satisfy their output distribution requirements.children=children.zip(requiredChildDistributions).map{case(child,distribution)ifchild.outputPartitioning.satisfies(distribution)=>childcase(child,BroadcastDistribution(mode))=>BroadcastExchangeExec(mode,child)case(child,distribution)=>valnumPartitions=distribution.requiredNumPartitions.getOrElse(conf.numShufflePartitions)ShuffleExchangeExec(distribution.createPartitioning(numPartitions),child)}//....}

这段代码的作用是：对当前物理算子operator的每个 child 进行检查，判断 child 的输出分布是否满足父算子的输入要求。如果不满足，就在已有 child 上方插入BroadcastExchangeExec或ShuffleExchangeExec。

operator.children

varchildren:Seq[SparkPlan]=operator.children

operator.children表示当前物理算子的输入子计划。

单 child 场景：

HashAggregateExec +- FileSourceScanExec

对HashAggregateExec来说：

children = Seq(FileSourceScanExec)

多 child 场景：

SortMergeJoinExec :- leftPlan +- rightPlan

对SortMergeJoinExec来说：

children = Seq(leftPlan, rightPlan)

叶子节点没有 child，例如FileSourceScanExec，其自身的children = Seq.empty。而ProjectExec、FilterExec这类一元算子通常会有一个 child。

operator.requiredChildDistribution

valrequiredChildDistributions:Seq[Distribution]=operator.requiredChildDistribution

requiredChildDistribution定义了当前物理算子要求所有 child 需要满足的数据分布Distribution。

requiredChildDistribution在SparkPlan中定义，所有物理算子都会包含该方法。

org.apache.spark.sql.execution.SparkPlan

defrequiredChildDistribution:Seq[Distribution]=Seq.fill(children.size)(UnspecifiedDistribution)

默认含义是：有几个 child，就返回几个UnspecifiedDistribution（对 child 的数据分区不关心）。

如果算子没有重写requiredChildDistribution，通常表示它对 child 数据分区没有特殊要求，比如ProjectExec、FilterExec这类一元算子。而常见的会发生 Shuffle 的算子，如HashAggregateExec、SortMergeJoinExec，会要求 child 满足一定的数据分布，因此会重写requiredChildDistribution。

常见 Shuffle 算子要求 child 满足的数据分布Distribution

不同算子对 child 的数据分布要求是不一样的，下面列举几种常见 Shuffle 算子所要求的数据分布。

① HashAggregateExec

HashAggregateExec的requiredChildDistribution方法在BaseAggregateExec中实现。

org/apache/spark/sql/execution/aggregate/BaseAggregateExec

overridedefrequiredChildDistribution:List[Distribution]={requiredChildDistributionExpressionsmatch{caseSome(exprs)ifexprs.isEmpty=>AllTuples::NilcaseSome(exprs)=>ClusteredDistribution(exprs)::NilcaseNone=>UnspecifiedDistribution::Nil}}

  • 有指定表达式且为空

    这种场景常见于全局聚合，如SELECT COUNT(*) FROM t，要求所有数据进入同一个分区，Distribution数据分布要求为AllTuples。

  • 指定了表达式且非空

    这种场景常见于普通分组聚合，如SELECT key, COUNT(*) FROM t GROUP BY key，要求 child 节点输出的数据按照这些表达式聚集或分区，数据分布要求为ClusteredDistribution。

  • 没有指定任何分布要求

对 child 节点的数据分布没有要求。

② SortMergeJoinExec 或 ShuffledHashJoinExec

对于 Shuffle Join，例如SortMergeJoinExec或ShuffledHashJoinExec，它们有两个 child，并且对左右 child 都有ClusteredDistribution数据分布要求（按照 Join key 做相同的分布）。

org/apache/spark/sql/execution/joins/ShuffledJoin

overridedefrequiredChildDistribution:Seq[Distribution]={ClusteredDistribution(leftKeys)::ClusteredDistribution(rightKeys)::Nil}

上面频繁出现Distribution这个概念，它到底是什么？下面一起看看。

Distribution

org/apache/spark/sql/catalyst/plans/physical/Distribution

sealedtraitDistribution{// 该 Distribution 要求的分区数量。如果是 None，表示任意分区数量都可以满足这个 Distribution。defrequiredNumPartitions:Option[Int]// 为该 Distribution 创建一个默认的 Partitioning。这个 Partitioning 既能满足当前 Distribution 的要求，又能匹配给定的分区数量defcreatePartitioning(numPartitions:Int):Partitioning}

Distribution表示“数据应该怎么分布”：当一个查询在多台机器上并行执行时，具有相同表达式值的元组应该如何分布。

前面涉及了不同算子所要求的Distribution，下面将介绍常见的Distribution类型：

  • UnspecifiedDistribution

不对 child 输入数据的分区方式提出要求，如普通投影、过滤。

caseobjectUnspecifiedDistributionextendsDistribution{overridedefrequiredNumPartitions:Option[Int]=NoneoverridedefcreatePartitioning(numPartitions:Int):Partitioning={thrownewIllegalStateException("UnspecifiedDistribution does not have default partitioning.")}}
  • AllTuples

只有一个分区的分布，表示整个数据集必须放在一个分区里，通常用于需要全局处理的操作。

caseobjectAllTuplesextendsDistribution{overridedefrequiredNumPartitions:Option[Int]=Some(1)overridedefcreatePartitioning(numPartitions:Int):Partitioning={assert(numPartitions==1,"The default partitioning of AllTuples can only have 1 partition.")SinglePartition}}
  • ClusteredDistribution

要求相同 key 的数据落在同一分区，通常用于 Group By、窗口函数。

caseclassClusteredDistribution(clustering:Seq[Expression],requiredNumPartitions:Option[Int]=None)extendsDistribution{require(clustering!=Nil,// 参与数据分布的表达式不能为空"The clustering expressions of a ClusteredDistribution should not be Nil.")overridedefcreatePartitioning(numPartitions:Int):Partitioning={assert(requiredNumPartitions.isEmpty||requiredNumPartitions.get==numPartitions,s"This ClusteredDistribution requires${requiredNumPartitions.get}partitions, but "+s"the actual number of partitions is$numPartitions.")HashPartitioning(clustering,numPartitions)}}

  • HashClusteredDistribution

    不仅要求相同 key 的数据在一起，还要求按 hash 规则进入指定分区，常用于 join 算子。相比ClusteredDistribution，它提供了更强的保障。

caseclassHashClusteredDistribution(expressions:Seq[Expression],requiredNumPartitions:Option[Int]=None)extendsDistribution{require(expressions!=Nil,// 参与数据分布的表达式不能为空"The expressions of a HashClusteredDistribution should not be Nil.")overridedefcreatePartitioning(numPartitions:Int):Partitioning={assert(requiredNumPartitions.isEmpty||requiredNumPartitions.get==numPartitions,s"This HashClusteredDistribution requires${requiredNumPartitions.get}partitions, but "+s"the actual number of partitions is$numPartitions.")HashPartitioning(expressions,numPartitions)}}
  • BroadcastDistribution

是一种广播分布要求，数据会被广播到每个执行节点上。

caseclassBroadcastDistribution(mode:BroadcastMode)extendsDistribution{overridedefrequiredNumPartitions:Option[Int]=Some(1)// 广播数据在被广播前通常会先收集成一个整体，然后作为一个广播变量发到各个 executor。所以从 Spark 物理分区语义上看，默认分区数要求是 1。overridedefcreatePartitioning(numPartitions:Int):Partitioning={assert(numPartitions==1,"The default partitioning of BroadcastDistribution can only have 1 partition.")BroadcastPartitioning(mode)}}

Partitioning

org/apache/spark/sql/catalyst/plans/physical/partitioning

traitPartitioning{// 该 SparkPlan 输出 RDD 的分区数目valnumPartitions:Int// 当前 child 的 Partitioning 是否能够满足下游要求的 Distribution，不满足时返回 false。// 要满足需要两个条件：① 分区数 numPartitions 要相等；// ② satisfies0 方法返回 true，包括两种情况：1. 对子节点的分布没有要求；2. 全局处理。finaldefsatisfies(required:Distribution):Boolean={required.requiredNumPartitions.forall(_==numPartitions)&&satisfies0(required)}// 1、如果 requiredChildDistribution 为 UnspecifiedDistribution，则说明对子节点的分布没有要求，返回 true；// 2、如果 requiredChildDistribution 为 AllTuples，则只要 numPartitions == 1，返回 true；// 3、其他情况，返回 false。// 具体的 Partitioning 类型会对 satisfies0 进行重写。protecteddefsatisfies0(required:Distribution):Boolean=requiredmatch{caseUnspecifiedDistribution=>truecaseAllTuples=>numPartitions==1case_=>false}}

Partitioning定义了一个物理算子输出数据的分区方式。

常见Partitioning类型有：

UnknownPartitioning：未知分区方式；

SinglePartition：单分区；

BroadcastPartitioning：广播分区；

HashPartitioning：基于哈希的分区方式；

RangePartitioning：基于范围的分区方式，通过确定分区键是否在某个范围内来选择分区；

Distribution 和 Partitioning 的关系

Distribution是“要求”，Partitioning是“实现”。

比如某个物理算子要求 child 中相同 key 的数据必须在同一个分区，这是一种Distribution要求。Spark 为了满足它，可能会创建HashPartitioning(keys, numPartitions)，这就是具体的Partitioning实现，也就是通过hash(key)的方式把数据分到指定数量的分区中。

简单总结：

Distribution = 算子需要什么样的数据分布 Partitioning = 当前数据实际是什么样的分区方式 Shuffle = 当现有 Partitioning 不能满足 Distribution 时，Spark 插入 Exchange 来重新分区

children.zip(requiredChildDistributions)

children=children.zip(requiredChildDistributions).map{case(child,distribution)ifchild.outputPartitioning.satisfies(distribution)=>childcase(child,BroadcastDistribution(mode))=>BroadcastExchangeExec(mode,child)case(child,distribution)=>valnumPartitions=distribution.requiredNumPartitions.getOrElse(conf.numShufflePartitions)ShuffleExchangeExec(distribution.createPartitioning(numPartitions),child)}

这一步会把当前算子的每个 child 和它要求的Distribution配对。

例如SortMergeJoinExec算子：

children: leftPlan rightPlan requiredChildDistributions: ClusteredDistribution(leftJoinKeys) ClusteredDistribution(rightJoinKeys)

配对后变成：

(leftPlan, ClusteredDistribution(leftJoinKeys)) (rightPlan, ClusteredDistribution(rightJoinKeys))

后续map会逐个检查每个 child 是否满足对应的Distribution。

第一分支：child 已经满足分布要求

case(child,distribution)ifchild.outputPartitioning.satisfies(distribution)=>child

如果 child 的输出分区方式已经满足父算子的分布要求，就直接返回原 child。例如，ProjectExec、FilterExec等物理算子并不要求 child 具备特定的输入分布。

第二分支：需要 BroadcastExchangeExec

case(child,BroadcastDistribution(mode))=>BroadcastExchangeExec(mode,child)

如果父算子要求 child 满足BroadcastDistribution，说明该 child 不需要普通 Shuffle，而是需要被广播。

典型场景是 Broadcast Hash Join：

SELECT/*+ BROADCAST(s) */*FROMbig_table bJOINsmall_table sONb.id=s.id;

对于小表s，父算子会要求：

BroadcastDistribution(HashedRelationBroadcastMode(keys))

第三分支：需要 ShuffleExchangeExec

case(child,distribution)=>valnumPartitions=distribution.requiredNumPartitions.getOrElse(conf.numShufflePartitions)ShuffleExchangeExec(distribution.createPartitioning(numPartitions),child)

如果 child 输出分区不满足当前算子的Distribution，且当前算子要求的不是 broadcast，就进入普通 Shuffle 分支。

1 确定 Shuffle 分区数

valnumPartitions=distribution.requiredNumPartitions.getOrElse(conf.numShufflePartitions)

如果distribution.requiredNumPartitions有值，则优先使用它。例如AllTuples.requiredNumPartitions = Some(1)，表示强制单分区。

如果没有强制分区数，则使用conf.numShufflePartitions。

org/apache/spark/sql/internal/SQLConf

defdefaultNumShufflePartitions:Int=getConf(SHUFFLE_PARTITIONS)defnumShufflePartitions:Int={if(adaptiveExecutionEnabled&&coalesceShufflePartitionsEnabled){getConf(COALESCE_PARTITIONS_INITIAL_PARTITION_NUM).getOrElse(defaultNumShufflePartitions)}else{defaultNumShufflePartitions}}

因此：

  • 未开启 AQE 时，shuffle 分区数等于spark.sql.shuffle.partitions。
  • 开启 AQE 且开启 Shuffle 分区合并时，初始分区数可能来自spark.sql.adaptive.coalescePartitions.initialPartitionNum；如果没有设置initialPartitionNum，仍然回退到spark.sql.shuffle.partitions。

相关参数：

①spark.sql.shuffle.partitions

该参数表示在进行 join 或聚合操作且需要 shuffle 数据时，默认使用的分区数量。默认值是 200。

valSHUFFLE_PARTITIONS=buildConf("spark.sql.shuffle.partitions").doc("The default number of partitions to use when shuffling data for joins or aggregations.").version("1.1.0").intConf.checkValue(_>0,"The value of spark.sql.shuffle.partitions must be positive").createWithDefault(200)

②spark.sql.adaptive.coalescePartitions.initialPartitionNum

该参数表示在进行分区合并之前，shuffle 分区的初始数量。

如果没有设置该值，则它等于spark.sql.shuffle.partitions。

这个配置只有在spark.sql.adaptive.enabled和spark.sql.adaptive.coalescePartitions.enabled都为true时才会生效。

valCOALESCE_PARTITIONS_INITIAL_PARTITION_NUM=buildConf("spark.sql.adaptive.coalescePartitions.initialPartitionNum").doc("The initial number of shuffle partitions before coalescing. If not set, it equals to "+s"${SHUFFLE_PARTITIONS.key}. This configuration only has an effect when "+s"'${ADAPTIVE_EXECUTION_ENABLED.key}' and '${COALESCE_PARTITIONS_ENABLED.key}' "+"are both true.").version("3.0.0").intConf.checkValue(_>0,"The initial number of partitions must be positive.").createOptional

spark.sql.adaptive.enabled：表示是否开启 Spark SQL 的 AQE，即 Adaptive Query Execution，自适应查询执行（默认关闭）。

spark.sql.adaptive.coalescePartitions.enabled：表示是否开启 AQE 下的 shuffle 分区合并功能（默认开启）。

2 创建具体的分区

distribution.createPartitioning(numPartitions)

创建满足当前算子要求的Distribution分布。

常见映射关系：

ClusteredDistribution(keys) -> HashPartitioning(keys, numPartitions) OrderedDistribution(ordering) -> RangePartitioning(ordering, numPartitions) AllTuples -> SinglePartition

3 创建 ShuffleExchangeExec

ShuffleExchangeExec(partitioning,child)

在当前物理算子的子节点child外面包一层ShuffleExchangeExec，强制对child的输出数据做一次 shuffle。在生成新的物理计划树时，当前物理算子的 child 会被替换为这个新的ShuffleExchangeExec，从而在当前算子和原计划之间增加一层 shuffle。

例如，一个聚合 SQL 生成的初始物理计划是：

HashAggregateExec(final) +- HashAggregateExec(partial) +- FileSourceScanExec

经过EnsureRequirements后：

HashAggregateExec(final) +- ShuffleExchangeExec(HashPartitioning(user_id, 200)) +- HashAggregateExec(partial) +- FileSourceScanExec

ShuffleExchangeExec 将分区数带入 Spark Core

ShuffleExchangeExec持有最终的输出分区方式：

org/apache/spark/sql/execution/exchange/ShuffleExchangeExec

caseclassShuffleExchangeExec(overridevaloutputPartitioning:Partitioning,child:SparkPlan,shuffleOrigin:ShuffleOrigin=ENSURE_REQUIREMENTS)extendsShuffleExchangeLike

其中outputPartitioning.numPartitions就是 SQL 物理计划层面已经确定的 Shuffle 分区数。

执行时会构造 Shuffle 依赖：

valshuffleDependency=ShuffleExchangeExec.prepareShuffleDependency(inputRDD,child.output,outputPartitioning,serializer,writeMetrics)

进入 Spark Core 后，分区数会体现在ShuffleDependency的partitioner上。

core/src/main/scala/org/apache/spark/Dependency.scala

classShuffleDependency[K:ClassTag,V:ClassTag,C:ClassTag](_rdd:RDD[_<:Product2[K,V]],valpartitioner:Partitioner,...)

其中partitioner.numPartitions决定了 Shuffle reduce 端的分区数，也通常对应 reduce task 的数量。

核心结论

Spark SQL Shuffle 分区数决定机制主流程可以概括为：

父算子声明 requiredChildDistribution -> EnsureRequirements 检查 child.outputPartitioning 是否满足要求 -> 满足则复用 child -> 要求 BroadcastDistribution 则插入 BroadcastExchangeExec -> 否则插入 ShuffleExchangeExec -> Shuffle 分区数优先取 distribution.requiredNumPartitions -> 没有强制要求时取 conf.numShufflePartitions -> conf.numShufflePartitions 通常来自 spark.sql.shuffle.partitions，默认 200

Spark SQL 的 Shuffle 分区数不是由单一位置决定的，而是由逻辑计划、物理计划、分布需求、用户显式设置和 AQE 共同决定。

分区数的优先级可以按下面理解：

  1. 显式重分区优先，例如df.repartition(n)、repartitionByRange(n, ...)，分区数直接使用用户指定的n。
  2. 算子强制分区数优先，例如AllTuples.requiredNumPartitions = Some(1)，会生成单分区 Shuffle。
  3. 普通 Join、Group By、Distinct、Window 等场景通常使用conf.numShufflePartitions，也就是spark.sql.shuffle.partitions。
  4. 开启 AQE 后，spark.sql.shuffle.partitions更多表示初始 Shuffle 分区数，最终 Shuffle Read task 数可能被合并或因倾斜拆分而变化。

参考

Spark SQL 源码研读系列 06：Executed Plan

标签： 网站建设 企业官网 项目流程 UI设计 前端开发

热门文章

文章分类

标签云

网站建设 响应式设计 用户体验 SEO优化 前端开发 小程序 电商平台 性能优化

相关文章

您可能感兴趣的其他内容

Active Directory 安全攻防（七）
2026/7/21 9:04:20 网站建设

Active Directory 安全攻防（七）

一、引言 在信息安全领域有一条基本原则：任何人如果获得了一台计算机未加密系统磁盘的访问权限，就等于拥有了这台计算机。这条原则适用于所有 Windows 系统，但当目标是域控制器时，其影响范围会被急剧放大。 对于域控制器而言，这条原则可以进一步延伸为： 任何人如果获取…...

阅读更多 →
TMS320F2807x CLA与DMA协同设计：从架构解析到电机控制实战
2026/7/21 9:03:18 网站建设

TMS320F2807x CLA与DMA协同设计：从架构解析到电机控制实战

1. 项目概述&#xff1a;为什么需要深入理解CLA与DMA的协同 在电机控制、数字电源或者任何对实时性有苛刻要求的嵌入式系统里&#xff0c;主CPU&#xff08;C28x&#xff09;常常被各种任务挤占得喘不过气。ADC采样、PWM更新、复杂的浮点运算、通信协议处理……这些任务如果全堆…...

阅读更多 →
Java面试突击：构建知识体系与场景题思维，高效备战2025秋招
2026/7/21 9:00:58 网站建设

Java面试突击：构建知识体系与场景题思维，高效备战2025秋招

如果你正在准备2025年秋季的Java校招或社招&#xff0c;并且感觉时间紧迫、知识点庞杂、无从下手&#xff0c;那么这篇文章就是为你准备的。市面上有太多“Java面试宝典”&#xff0c;动辄几百页PDF&#xff0c;但真正能让你在短期内抓住重点、建立体系、应对实战的却很少。本文…...

阅读更多 →

需要专业的网站建设服务？

联系我们获取免费的网站建设咨询和方案报价，让我们帮助您实现业务目标

立即咨询