PySpark/Iceberg分桶Join未消除Exchange步骤的原因排查

我尝试在PySpark中使用Iceberg格式对两张表执行Join操作，希望通过分桶（bucketing）提升性能并避免Shuffle，但似乎完全没有效果。为什么分桶操作没有移除Exchange步骤？

非分桶Join代码： #

db = "ml_recommendations" 
col = "id" 

df1 = spark.range(1, 10 ** 6)  # 创建单列"id"的DF
df1.writeTo(f"{db}.no_bucket_table1") \ 
    .using('iceberg') \
    .option("fanout-enabled", "true") \
    .createOrReplace() 

df2 = spark.range(1, 10 ** 5)  # 创建单列"id"的DF
df2.writeTo(f"{db}.no_bucket_table2") \ 
    .using('iceberg') \ 
    .option("fanout-enabled", "true") \ 
    .createOrReplace() 

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1) 

df = spark.sql( 
    f"select * from {db}.no_bucket_table1 t1 " 
    f"inner join {db}.no_bucket_table2 t2 " 
    f"on t1.{col}=t2.{col}" 
) 
df.explain(extended=True)

非分桶Join执行结果： #

== Optimized Logical Plan ==
Join Inner, (id#8L = id#9L)
:- Filter isnotnull(id#8L)
:  +- RelationV2[id#8L] Iceberg iceberg.ml_recommendations.no_bucket_table1
+- Filter isnotnull(id#9L)
   +- RelationV2[id#9L] Iceberg iceberg.ml_recommendations.no_bucket_table2
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [id#8L], [id#9L], Inner
   :- Sort [id#8L ASC NULLS FIRST], false, 0
   :  +- **Exchange hashpartitioning**(id#8L, 400), ENSURE_REQUIREMENTS, [plan_id=201]
   :     +- Filter isnotnull(id#8L)
   :        +- BatchScan[id#8L] iceberg.ml_recommendations.no_bucket_table1 (branch=null) [filters=id IS NOT NULL, groupedBy=] RuntimeFilters: []
   +- Sort [id#9L ASC NULLS FIRST], false, 0
      +- **Exchange hashpartitioning**(id#9L, 400), ENSURE_REQUIREMENTS, [plan_id=202]
         +- Filter isnotnull(id#9L)
            +- BatchScan[id#9L] iceberg.ml_recommendations.no_bucket_table2 (branch=null) [filters=id IS NOT NULL, groupedBy=] RuntimeFilters: []

分桶Join代码： #

df1 = spark.range(1, 10 ** 6)  # 创建单列"id"的DF
df1.writeTo(f"{db}.bucket_table1") \ 
    .using('iceberg') \ 
    .option("fanout-enabled", "true") \ 
    .partitionedBy(pf.bucket(100, "id")) \ 
    .createOrReplace() 

df2 = spark.range(1, 10 ** 5)  # 创建单列"id"的DF
df2.writeTo(f"{db}.bucket_table2") \ 
    .using('iceberg') \ 
    .option("fanout-enabled", "true") \ 
    .partitionedBy(pf.bucket(100, "id")) \ 
    .createOrReplace() 

df = spark.sql( 
    f"select * from {db}.bucket_table1 t1 " 
    f"inner join {db}.bucket_table2 t2 " 
    f"on t1.{col}=t2.{col}" 
) 
df.explain(extended=True)

分桶Join执行结果： #

== Optimized Logical Plan ==
Join Inner, (id#43L = id#44L)
:- Filter isnotnull(id#43L)
:  +- RelationV2[id#43L] Iceberg iceberg.ml_recommendations.bucket_table1
+- Filter isnotnull(id#44L)
   +- RelationV2[id#44L] Iceberg iceberg.ml_recommendations.bucket_table2
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [id#43L], [id#44L], Inner
   :- Sort [id#43L ASC NULLS FIRST], false, 0
   :  +- **Exchange hashpartitioning**(id#43L, 400), ENSURE_REQUIREMENTS, [plan_id=422]
   :     +- Filter isnotnull(id#43L)
   :        +- BatchScan[id#43L] iceberg.ml_recommendations.bucket_table1 (branch=null) [filters=id IS NOT NULL, groupedBy=] RuntimeFilters: []
   +- Sort [id#44L ASC NULLS FIRST], false, 0
      +- **Exchange hashpartitioning**(id#44L, 400), ENSURE_REQUIREMENTS, [plan_id=423]
         +- Filter isnotnull(id#44L)
            +- BatchScan[id#44L] iceberg.ml_recommendations.bucket_table2 (branch=null) [filters=id IS NOT NULL, groupedBy=] RuntimeFilters: []

1. Iceberg分桶与Spark原生分桶的差异 #

你使用的pf.bucket(100, "id")是Iceberg自身的分桶策略，属于表的分区配置，而非Spark原生的分桶（bucketBy）。Spark默认不会识别Iceberg的分桶元数据，因此无法触发分桶Join优化，仍会走常规的SortMergeJoin并执行Shuffle。

2. 缺少Iceberg分桶Join的启用配置 #

要让Spark识别Iceberg的分桶信息并跳过Shuffle，必须开启以下配置：

# 开启Iceberg分桶Join支持
spark.conf.set("spark.sql.iceberg.bucketedJoin.enabled", "true")
# 确保未排除相关优化规则
spark.conf.set("spark.sql.optimizer.excludedRules", "")

3. 分桶一致性验证 #

两张表的分桶数量必须完全一致，且Join键必须与分桶键完全匹配。你的代码中两张表都使用100个分桶、分桶键为id，这部分是符合要求的，但如果缺少上述配置，Spark仍不会利用分桶信息。

4. 优化生效的验证方法 #

开启配置后重新执行Join，查看物理计划：

• 成功触发分桶Join时，Physical Plan会显示BucketedJoin而非SortMergeJoin，且不会出现Exchange hashpartitioning步骤；
• BatchScan的groupedBy字段会显示分桶信息，例如groupedBy=hash(id, 100)。

额外注意事项 #

• 确保使用的Iceberg版本≥0.13，该版本开始支持分桶Join优化；
• 不要混淆Iceberg的partitionedBy(pf.bucket(...))和Spark的bucketBy(...)，前者是Iceberg层的分区策略，后者是Spark原生分桶，两种方式都可以配合上述配置实现分桶Join，但推荐使用Iceberg自身的分桶策略。

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