我试图了解Distribute by 子句以及如何在Spark-SQL 中使用它来优化Sort-Merge Joins。
据我了解，Spark Sql 优化器将根据连接键（shuffle 阶段）分配两个参与表（连接）的数据集，以将相同的键放在同一分区中。如果是这样，那么如果我们在 sql 中使用distribute by，那么我们也在做同样的事情。
那么distribute by 可以通过什么方式来改善连接性能？还是在加载过程中将数据写入磁盘时最好使用distribute by，以便后续使用该数据的查询将受益于它而不必洗牌？
您能否通过一个真实示例来解释在 Spark-SQL 中使用 distribute by/cluster by 调整连接？
【问题讨论】：
标签： python apache-spark apache-spark-sql pyspark-sql
让我试着回答你问题的每个部分：
据我了解，Spark Sql 优化器将根据连接键（shuffle 阶段）分配两个参与表（连接）的数据集，以将相同的键放在同一分区中。如果是这样的话，那么如果我们在sql中使用distribute by，那么我们也在做同样的事情。
是的，没错。
那么通过什么方式分发可以改善连接性能？
有时您的某个表已经分布，例如该表已被分桶或数据在连接之前通过相同的键聚合。在这种情况下，如果您还显式地重新分区第二个表（分布），您将在连接的两个分支中实现相同的分区，并且 Spark 将不会在第一个分支中引发更多的随机播放（有时这被称为单边随机播放-free 加入，因为 shuffle 只会发生在加入的一个分支中 - 您调用 repartition / 分发的分支）。另一方面，如果您没有明确地重新分区另一个表，Spark 将看到连接的每个分支都有不同的分区，因此它将打乱两个分支。因此，在某些特殊情况下，调用 repartition (distribute by) 可以为您节省一次 shuffle。
请注意，要完成这项工作，您还需要在两个分支中实现相同数量的分区。因此，如果您有两个要在键 user_id 上加入的表，并且如果第一个表使用该键被分桶到 10 个桶中，那么您需要将另一个表也重新分区为 10 个分区，使用相同的键，然后加入将只有一次 shuffle（在物理计划中，您可以看到仅在 join 的一个分支中会有 Exchange 运算符）。
或者在加载过程中将数据写入磁盘时最好使用distribute by，这样使用该数据的后续查询将受益于它，而不必对其进行洗牌？
嗯，这实际上称为分桶（按集群方式），它允许您预洗数据一次，然后每次读取数据并通过分桶时使用的相同键加入（或聚合）数据，它将没有洗牌。所以是的，这是一种非常常见的技术，您只需在保存数据时支付一次费用，然后在每次读取数据时利用它。
【讨论】：