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

Question

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

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 应该使用它的第一个元素作为分区键。有没有办法 "force" Spark 合并 2 个 PairRDD 以使用相同的分区键？

Answer 1

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

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

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

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

Answer 2

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

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

In Apache Spark, why does RDD.union not preserve the partitioner?

partitioning

hadoop-partitioning

apache-spark