Spark配置spark.default.parallelism困惑：实际行为与文档不符

官方文档描述 #

spark.default.parallelism: 用户未指定时，join、reduceByKey、parallelize等转换返回的RDD默认分区数

spark.default.parallelism: 对于reduceByKey、join这类分布式shuffle操作，取父RDD中最大的分区数

实际测试与文档描述不符的场景 #

1. DataFrame场景 #

创建50分区的df1与3分区的df2进行inner join，按文档预期结果分区数应为50，但实际返回3：

df1 = spark.range(1,10000).repartition(50)
df2 = spark.range(1,100000).repartition(3)
df3 = df1.join(df2, df1.id == df2.id, 'inner')
df3.rdd.getNumPartitions() # 返回3

2. RDD场景 #

将上述两个DataFrame对应的RDD进行join，按文档预期结果分区数应为50，但实际返回53：

df1 = spark.range(1,10000).repartition(50)
df2 = spark.range(1,100000).repartition(3)
df3 = df1.rdd.join(df2.rdd)
df3.getNumPartitions() # 返回53

对spark.default.parallelism的正确理解 #

1. 作用范围：仅针对原生RDD API #

spark.default.parallelism是专门为RDD API设计的配置，DataFrame/DataSet API的shuffle分区逻辑完全独立，由spark.sql.shuffle.partitions（默认值200）控制，二者没有直接关联。

2. RDD shuffle操作的分区数规则 #

对于RDD的join、reduceByKey等shuffle操作，默认分区数的判定逻辑优先级如下：

• 优先使用操作显式指定的分区数；
• 若未指定：
  • 若参与操作的RDD有统一兼容的分区器（比如两个RDD使用相同分区数的HashPartitioner），则复用该分区器的分区数；
  • 若只有一个RDD有分区器，使用该RDD的分区数；
  • 若没有分区器或分区器不兼容，则使用spark.default.parallelism的配置值。

你测试中RDD join返回53个分区，说明当前环境的spark.default.parallelism值为53（通常由集群总核心数、本地模式CPU核心数等决定），且两个RDD未触发分区器复用逻辑，因此Spark使用了该配置值作为shuffle后的分区数。

3. DataFrame join的分区数逻辑 #

DataFrame的join结果分区数与spark.default.parallelism无关，主要由以下因素决定：

• 广播优化：当Spark判定其中一个表为小表（阈值由spark.sql.autoBroadcastJoinThreshold控制）时，会将小表广播到所有节点，无需shuffle大表，最终结果的分区数等于大表的分区数。你的测试中df1数据量远小于df2，触发了该优化，因此结果分区数等于df2的3个分区；
• shuffle类join：若未触发广播优化（比如两张表都很大），执行sort-merge join时，分区数由spark.sql.shuffle.partitions配置控制。

4. 文档描述的局限性 #

官方文档的描述简化了实际逻辑，未明确区分RDD与DataFrame的配置边界，也未提及分区器、广播优化等关键影响因素，这是导致测试结果与预期不符的核心原因。

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