Spark 的 Shuffle Sort Merge Join。一个 DataFrame 被分桶。 Spark 是否利用了这一点？

Question

我记得在使用 RDD 时，如果一个键值 RDD (rdd1) 具有已知的分区，那么使用不同的未分区键值 RDD (rdd2) 执行连接将带来性能优势。这是因为 1) 只有 rdd2 的数据需要通过网络传输，2) 通过将 rdd1 的键的分区应用于rdd2的key

我正在学习使用 DataFrames 的 Shuffle Sort Merge Joins。我正在阅读的书（Learning Spark，第 2 版）中的示例是用于连接两个基于 user_id 列的 DataFrame。该示例试图演示从联接操作中消除 Exchange 阶段，因此，在联接之前，两个 DataFrame 都按要联接的列分桶到相同数量的桶中。

我的问题是，如果只有一个 DataFrame 被分桶，会发生什么？显然，交换阶段将重新出现。但是如果我们知道 DataFrame1 被我们想要加入的列分桶到 N 个桶中，Spark 是否会使用这个分桶信息在网络上有效地传输 DataFrame2 的行，就像在 RDD 的情况下一样？ Spark 会将 DataFrame1 的行保留在原处，并且只对 DataFrame2 应用相同的分桶吗？ （假设 N 个桶导致分区中有合理数量的数据被执行程序加入）或者，Spark 是否低效地洗牌两个 DataFrame？

特别是，我可以想象这样一种情况，即我有一个“主”数据帧，我需要针对该数据帧与同一列上的其他补充数据帧执行许多独立连接。当然，只需要预先存储主 DataFrame 才能看到所有连接的性能优势吗？ （虽然我认为不厌其烦地存储补充 DataFrames 也不会受到伤害）

Answer 1

https://kb.databricks.com/data/bucketing.html 这解释了这一切，并在我总结的原始帖子上进行了一些修饰。

底线：

val t1 = spark.table("unbucketed")
val t2 = spark.table("bucketed")
val t3 = spark.table("bucketed")

Unbucketed - 分桶连接。两边都需要重新分区。

t1.join(t2, Seq("key")).explain()

使用重新分区取消分桶 - 分桶连接。未分包的一面是 正确重新分区，只需要一次随机播放。

t1.repartition(16, $"key").join(t2, Seq("key")).explain()

使用不正确的重新分区取消分桶（默认（200） - 分桶连接。 未分桶的一侧被错误地重新分区，并且两次洗牌 需要。

t1.repartition($"key").join(t2, Seq("key")).explain()

bucketed - 分桶连接。理想情况，双方相同 分桶，并且不需要洗牌。

t3.join(t2, Seq("key")).explain()

因此，双方都需要相同的分桶以获得最佳性能。