无分桶策略下，同分区两表Join如何避免Shuffle？

问题背景 #

我有两个Hive表A和B，具备以下特征：

• 拥有相同的分区字段：partition_1、partition_2
• 存在额外的id字段，且该字段在分区内未排序

在PySpark中执行内连接的代码如下：

df_A = spark.table("db.A")
df_B = spark.table("db.B")
df = df_A.join(df_B, how="inner", on=["partition_1", "partition_2", "id"])

执行计划中始终出现Shuffle操作：

+- == Initial Plan ==
  Project (23)
  +- SortMergeJoin Inner (22)
     :- Sort (18)
     :  +- Exchange (17)
     :     +- Filter (16)
     :        +- Scan parquet db.A (15)
     +- Sort (21)
        +- Exchange (20)
           +- Filter (19)
              +- Scan parquet db.B (7)

当创建采用分桶策略的相似表后：

df.write.partitionBy("partition_1", "partition_2").bucketBy(10, "id").saveAsTable(...)

Join操作不再出现Shuffle：

+- == Initial Plan ==
  Project (17)
  +- SortMergeJoin Inner (16)
     :- Sort (13)
     :  +- Filter (12)
     :     +- Scan parquet db.A (11)
     +- Sort (15)
        +- Filter (14)
           +- Scan parquet db.B (5)

我的疑问：

1. 能否无需重新创建分桶表，即可避免Join时的Shuffle？
2. 该Shuffle是针对全量数据执行，还是会考虑同分区情况进行优化？

我已尝试以下操作，但Shuffle仍存在：

• 在Join前对两表按分区字段重分区：df.repartition("partition_1", "partition_2")
• 在Join前对两表按分区和id字段重分区：df.repartition(numPartitions, "partition_1", "partition_2", "id")
• 在Join前按id字段排序

测试环境覆盖Databricks和EMR，表现一致。

解答 #

1. 无需创建分桶表能否避免Join阶段的Shuffle？ #

无法完全避免shuffle操作，但可以将shuffle提前到数据准备阶段，让Join本身不再触发新的Shuffle。

原因在于：

• 原始表仅按partition_1、partition_2分区，id在分区内无序，且不同分区的id分布没有对齐。Spark的SortMergeJoin要求两边数据在Join键（partition_1, partition_2, id）上满足两个条件：
  1. 数据按Join键的哈希值划分到相同数量的分区
  2. 每个分区内的数据按Join键排序
• 你之前尝试的repartition操作虽然能按Join键重新分区，但Spark不会将这种临时分区信息持久化到查询计划的元数据中，因此Join阶段仍会触发Shuffle来验证分区一致性。

正确的做法是，在Join前显式将数据组织成符合SortMergeJoin要求的结构：

# 指定与集群资源匹配的分区数（比如100，可根据实际调整）
num_partitions = 100

# 对表A按Join键重分区，并在分区内按Join键排序
df_A = spark.table("db.A") \
    .repartition(num_partitions, "partition_1", "partition_2", "id") \
    .sortWithinPartitions("partition_1", "partition_2", "id")

# 对表B执行完全相同的操作，确保分区数和分区器一致
df_B = spark.table("db.B") \
    .repartition(num_partitions, "partition_1", "partition_2", "id") \
    .sortWithinPartitions("partition_1", "partition_2", "id")

# 此时Join不会触发新的Shuffle
df = df_A.join(df_B, how="inner", on=["partition_1", "partition_2", "id"])

注意：这个方案只是把Join阶段的Shuffle提前到了repartition步骤，本质上还是存在一次Shuffle操作——因为原始数据的分布不符合Join要求，必须通过Shuffle重新组织数据。只有当数据本身就按Join键分区且有序时，才能完全跳过Shuffle。

2. Shuffle是否会考虑同分区情况优化？ #

Spark的Shuffle会基于Join键的整体进行优化，自然包含分区字段的逻辑：

• 由于你的Join键包含partition_1和partition_2，Shuffle时会先对这两个字段哈希，再对id哈希，最终相同partition_1+partition_2+id组合的行会被分配到同一个Shuffle分区。
• 这意味着Shuffle不会跨partition_1+partition_2的分区混洗数据，只会在每个分区内部对id相关的数据进行重新分配。如果查询中有分区过滤条件（比如where partition_1='xxx'），Spark还会先过滤掉无关分区，只对剩余分区的数据执行Shuffle。

简单来说：Shuffle是针对过滤后的有效数据执行，但会按Join键的分区字段+id做逻辑分组，不会无差别混洗全量数据。

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