为何分桶排序一致的Spark表Join时仍需重排序？

我在执行两个表的Full Outer Join操作，SQL语句如下：

SELECT
    a.user_id
FROM
    table1 a
    FULL OUTER JOIN
    table2 b
        ON a.user_id = b.user_id

table1和table2都是通过以下代码按user_id分桶（分桶数4）并排序后写入的：

df.write
    .mode("overwrite")
    .format("parquet")
    .bucketBy(4, "user_id")
    .sortBy("user_id")
    .option("path", "some/path/to/data/")
    .option("compression", "snappy")
    .saveAsTable("table1")

但查看物理执行计划时，发现Spark在FileScan之后仍然执行了Sort步骤，这不符合预期：

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [user_id#4483L]
   +- SortMergeJoin [user_id#4483L], [user_id#4485L], FullOuter
      :- Sort [user_id#4483L ASC NULLS FIRST], false, 0
      :  +- FileScan parquet default.table1[user_id#4483L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[s3://example/path_to_table1], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<user_id:bigint>, SelectedBucketsCount: 4 out of 4
      +- Sort [user_id#4485L ASC NULLS FIRST], false, 0
         +- FileScan parquet default.table2[user_id#4485L] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex(1 paths)[s3://example/path_to_table2], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<user_id:bigint>, SelectedBucketsCount: 4 out of 4

此前未分桶排序时，执行计划包含Exchange步骤，分桶后已经消除了该步骤，但希望同时消除Sort步骤。

需要在写入表前，先通过repartition将数据按分桶数和分桶键重新分区，修改后的写入代码如下：

df.write
    .repartition(4, col("user_id"))
    .mode("overwrite")
    .format("parquet")
    .bucketBy(4, "user_id")
    .sortBy("user_id")
    .option("path", "some/path/to/data/")
    .option("compression", "snappy")
    .saveAsTable("table1")

原理说明 #

bucketBy仅保证数据按指定规则分桶，但如果写入前的DataFrame分区数与分桶数不一致，Spark无法确保每个桶内的数据是严格全局有序的（可能存在同一桶的数据分散在多个写入分区中，最终合并到桶文件时顺序被打乱）。提前用repartition(4, col("user_id"))将数据分区数与分桶数对齐，且按分桶键分区，再配合sortBy，能保证每个写入分区对应一个桶，且桶内数据完全有序。这样Spark在读取时就能识别到桶内数据的有序性，从而跳过SortMergeJoin前的Sort步骤。

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