PySpark join 对共同分区的 RDD 进行洗牌

Question

from pyspark import SparkContext

sc = SparkContext()

rdd1 = sc.parallelize([('a', 1), ('b', 2), ('c', 3), ('d', 4)], numSlices=8)
rdd2 = rdd1.mapValues(lambda x: x)

这些 RDD 具有相同的分区：

rdd1.keys().glom().collect()
>>> [[], ['a'], [], ['b'], [], ['c'], [], ['d']]

rdd2.keys().glom().collect()
>>> [[], ['a'], [], ['b'], [], ['c'], [], ['d']]

这里有多个答案表明加入共同分区的数据不会导致洗牌，这对我来说很有意义。示例：Does a join of co-partitioned RDDs cause a shuffle in Apache Spark?

但是，当我使用 PySpark 加入这些共同分区的 RDD 时，数据会被打乱到一个新的分区中：

rdd1.join(rdd2).keys().glom().collect()
>>> [['a'], [], ['c'], ['b'], [], ['d'], [], [], [], [], [], [], [], [], [], []]

即使我将新分区的数量设置为原来的 8 个，分区也会发生变化：

rdd1.join(rdd2, numPartitions=8).keys().glom().collect()
>>> [['a'], [], ['c'], ['b'], [], ['d'], [], []]

为什么我无法避免使用这些共同分区的 RDD 进行洗牌？

我使用的是 Spark 1.6.0。

Answer 1

在这种情况下，rdd1 和 rdd2 都没有被分区

rdd1 = sc.parallelize([('a', 1), ('b', 2), ('c', 3), ('d', 4)])
rdd2 = rdd1.mapValues(lambda x: x)

rdd1.partitioner is None
## True

rdd2.partitioner is None
# True

所以根据定义没有共同分区。虽然您可以对数据进行分区并加入：

n = rdd1.getNumPartitions()
rdd1part = rdd1.partitionBy(n)
rdd2part = rdd2.partitionBy(n)

rdd1part.join(rdd2part)  # rdd1part and rdd2part are co-partitioned

这只会重新排列 DAG，不会阻止 shuffle。

另见Default Partitioning Scheme in Spark