控制Spark Join哈希分区：减少重复哈希分区的可行性探究

默认情况下Spark确实会为两次Join分别对A按[a,b]、[a,c]做哈希分区，导致A被Shuffle两次。要只按公共列a完成三次分区（A、B、C各一次），核心思路是提前对三个DataFrame按a做哈希分区并持久化，让后续Join复用已有分区，避免重复Shuffle。

具体实现步骤： #

1. 统一按a哈希分区 #

用repartition指定按a列做哈希分区，分区数建议和Spark默认的spark.sql.shuffle.partitions保持一致（默认200）：

# 按a列哈希分区，对齐默认并行度
A_part = A.repartition("a")
B_part = B.repartition("a")
C_part = C.repartition("a")

2. 持久化分区后的数据集 #

为了避免分区逻辑重复执行，把分区后的DataFrame缓存起来：

# 选择合适的存储级别，这里用内存+磁盘兜底
A_part.cache()
B_part.cache()
C_part.cache()

# 手动触发缓存（也可以在Join时自动触发，提前触发能让后续Join更快）
A_part.count()
B_part.count()
C_part.count()

3. 基于分区后的数据集执行Join #

此时执行Join时，Spark会复用已有的按a分区的结构，不需要再对A做两次全量Shuffle——因为同一a值的数据已经在同一个分区里，只需要在分区内按b或c做匹配（如果是SortMergeJoin的话，会在分区内排序，而非全量Shuffle）：

AB = A_part.join(B_part, ["a", "b"], "inner")
AC = A_part.join(C_part, ["a", "c"], "inner")

# 查看优化后的物理计划
AB.explain()
AC.explain()

优化后的物理计划变化 #

对比原来的计划，你会发现A_part不再出现两次Exchange hashpartitioning(a, b)和Exchange hashpartitioning(a, c)，而是直接从缓存的分区数据集读取，只在分区内做排序和Join操作，节省了一次针对A的Shuffle。

另外，开启Spark自适应执行（Adaptive Execution）会进一步辅助优化，但手动提前分区+持久化是最可控的方式。

注意事项： #

• 分区数不要随意设置，建议对齐spark.sql.shuffle.partitions，避免分区过多或过少影响性能
• 如果数据集极大，可以用persist(StorageLevel.MEMORY_AND_DISK_SER)来序列化存储，减少内存占用

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