Spark分桶表Join前出现排序的原因及移除排序阶段的方法

首先，咱们得搞清楚为什么会出现这个“无shuffle但有排序”的情况——其实核心问题就是Spark没意识到你的分桶表内部已经是按UserID有序的，所以才会画蛇添足地加一个本地排序步骤。下面具体拆解原因和解决办法：

为什么会出现额外排序？ #

1. 元数据没记录桶内排序信息
   你写表时用了sortBy("UserID")，但默认情况下，Spark（尤其是2.x早期版本）不会把这个“桶内有序”的信息存到表的元数据里。当Spark执行Sort Merge Join（大表Join的默认策略）时，它要求两边数据必须按Join键有序，既然看不到元数据里的排序标记，就会认为桶内数据是乱序的，于是在Join前做一次本地排序（因为数据已经在正确的桶里，所以不需要shuffle，只是桶内排序）。

2. Spark版本对分桶排序的支持不足
   在Spark 2.3之前，官方对分桶表的排序元数据支持并不完善，即使你写表时指定了sortBy，也不会自动同步到 metastore 里，导致读取时完全识别不到有序性。

怎么移除这个多余的排序阶段？ #

针对上面的原因，咱们一步步解决：

1. 写入时强制记录排序元数据 #

在写入分桶表前，先开启spark.sql.sources.bucketing.sorting.enabled配置，让Spark把sortBy的列信息写入表的元数据：

// 先设置配置
spark.conf.set("spark.sql.sources.bucketing.sorting.enabled", "true")
// 再写表
df.write.bucketBy(200, "UserID").sortBy("UserID").saveAsTable("topn_bucket_test")

写完之后，你可以用DESCRIBE EXTENDED topn_bucket_test查看元数据，确认里面有没有Sort Columns: [UserID]的条目——有了这个，Spark读取时就知道桶内已经有序了。

2. 读取时提示Spark用Bucketed Join #

如果元数据已经正确，但Spark还是没识别到，可以在Join时加hint("bucket")，强制Spark使用Bucketed Join策略，直接利用分桶结构跳过排序：

val bucketedTable = spark.table("topn_bucket_test")
val t2 = spark.table("t2")
// 给Join加bucket提示
bucketedTable.join(t2.hint("bucket"), "UserID")

3. 确保Spark版本和配置正确 #

• 尽量升级到Spark 2.3+（推荐3.x），新版本对分桶表的元数据支持更完善。
• 如果是Hive metastore管理的表，确保spark.sql.hive.convertMetastoreParquet配置为true（默认就是true，不用改，但如果被手动关了要打开），这样Spark能正确读取Hive里的分桶排序元数据。

4. 验证效果 #

做完上面的操作后，再看执行计划，你会发现那个多余的Sort阶段就消失了——因为Spark现在明确知道分桶表的每个桶里已经按UserID排好序，直接就能进行Sort Merge Join，不需要额外排序。

内容的提问来源于stack exchange，提问作者Rajesh Kumar Dash