在 Apache Spark 中，为什么 RDD.union 不保留分区程序？答案

【问题标题】：In Apache Spark, why does RDD.union not preserve the partitioner?在 Apache Spark 中，为什么 RDD.union 不保留分区程序？
【发布时间】：2015-07-10 17:36:30
【问题描述】：

众所周知，Spark 中的分区器对任何“宽”操作都有巨大的性能影响，因此它通常在操作中进行自定义。我正在试验以下代码：

val rdd1 =
  sc.parallelize(1 to 50).keyBy(_ % 10)
    .partitionBy(new HashPartitioner(10))
val rdd2 =
  sc.parallelize(200 to 230).keyBy(_ % 13)

val cogrouped = rdd1.cogroup(rdd2)
println("cogrouped: " + cogrouped.partitioner)

val unioned = rdd1.union(rdd2)
println("union: " + unioned.partitioner)

我看到默认情况下cogroup() 总是生成带有自定义分区器的 RDD，但 union() 没有，它总是会恢复为默认值。这是违反直觉的，因为我们通常假设 PairRDD 应该使用它的第一个元素作为分区键。有没有办法“强制”Spark 合并 2 个 PairRDD 以使用相同的分区键？

【问题讨论】：

标签： apache-spark partitioning hadoop-partitioning

【解决方案1】：

union 是一种非常高效的操作，因为它不会移动任何数据。如果rdd1 有 10 个分区，rdd2 有 20 个分区，那么rdd1.union(rdd2) 将有 30 个分区：两个 RDD 的分区彼此相连。这只是记账变更，没有洗牌。

但它必然会丢弃分区器。为给定数量的分区构造一个分区器。生成的 RDD 有许多不同于rdd1 和rdd2 的分区。

联合后可以运行repartition来打乱数据并按键组织。

上述有一个例外。如果rdd1 和rdd2 具有相同的分区器（具有相同数量的分区），则union 的行为不同。它将成对连接两个 RDD 的分区，使其具有与每个输入相同数量的分区。这可能涉及移动数据（如果分区没有位于同一位置），但不会涉及洗牌。在这种情况下，分区器被保留。（代码在PartitionerAwareUnionRDD.scala。）

【讨论】：

实际上有一个支持分区器的联合 RDD，我认为应该在可以保留分区的情况下自动使用它；不知道为什么它不在这里应用。见github.com/apache/spark/blob/… 和github.com/apache/spark/blob/master/core/src/main/scala/org/…
哇，酷！从来不知道那件事。看起来它仅在两个 RDD 具有相同的分区器时使用。我会把它添加到答案中，谢谢！
非常感谢！这是一个非常重要的优化。顺便说一句，如果这不是所有情况下的最佳选择，我总是可以编写一个 zip + 分区内联合
几乎所有东西都使用 HashPartitioner。因此，如果您的 DataFrame 具有相同数量的分区，我希望这就足够了。您只需打印 df.partitioner 和 df.partitions 即可查看发生了什么。
只是补充一点，正确的命令是df.rdd.partitioner 和df.rdd.getNumPartitions。知道为什么我的 DF 没有任何分区器（无），即使我正在重新分区它们？

【解决方案2】：

这不再是真的。如果两个 RDD 具有完全相同的分区器和分区数，则unioned RDD 也将具有相同的分区。这是在 https://github.com/apache/spark/pull/4629 中引入的，并被合并到 Spark 1.3 中。

【讨论】：