Spark-SQL中DISTRIBUTE BY和Shuffle的区别

Question

我试图理解Distribute by子句以及如何使用它来Spark-SQL优化Sort-Merge Joins。

根据我的理解，Spark Sql 优化器将根据连接键（洗牌阶段）分配（连接的）两个参与表的数据集，以将相同的键共同定位在同一分区中。如果是这样的话，那么如果我们distribute by在 sql 中使用 the ，那么我们也在做同样的事情。

那么可以用什么方法distribute by来改善连接性能呢？或者，最好distribute by在通过加载过程将数据写入磁盘时使用，以便使用该数据的后续查询将受益于它，而不必对其进行洗牌？

您能用一个真实的例子来解释一下distribute by/cluster by在 Spark-SQL 中使用调整连接吗？

Answer 1

让我尝试回答你问题的每一部分：

根据我的理解，Spark Sql 优化器将根据连接键（洗牌阶段）分配（连接的）两个参与表的数据集，以将相同的键共同定位在同一分区中。如果是这样的话，那么如果我们在sql中使用distribute by，那么我们也在做同样的事情。

对，那是正确的。

那么分布式可以通过什么方式来改善连接性能呢？

有时，您的表之一已经是分布式的，例如该表已分桶或在连接之前通过相同的键聚合了数据。在这种情况下，如果您还显式地重新分区第二个表（分配依据），您将在连接的两个分支中实现相同的分区，并且 Spark 不会在第一个分支中引发任何更多的洗牌（有时这被称为单侧洗牌） -free join 因为随机播放只会发生在 join 的一个分支中 - 您在其中调用 repartition/distribution by 的分支）。另一方面，如果您没有显式地重新分区另一个表，Spark 将看到连接的每个分支都有不同的分区，因此它将对两个分支进行混洗。所以在某些特殊情况下调用repartition（distribute by）可以节省一次shuffle。

请注意，要实现此目的，您还需要在两个分支中实现相同数量的分区。因此，如果您有两个表想要在该键上连接user_id，并且第一个表使用该键分为 10 个存储桶，那么您需要使用相同的键将另一个表也重新分区为 10 个分区，然后连接将只具有一次洗牌（在物理计划中您可以看到只有在连接的一个分支中才会有 Exchange 运算符）。

或者，在加载过程将数据写入磁盘时最好使用 Distribution by，这样使用该数据的后续查询将受益于它，而不必对其进行混洗？

嗯，这实际上称为分桶（聚类依据），它允许您预先对数据进行一次洗牌，然后每次您读取数据并通过与分桶相同的键将其连接（或聚合）时，它将免洗牌。所以，是的，这是一种非常常见的技术，您在保存数据时只需支付一次成本，然后在每次读取数据时都可以利用它。