PySpark加入混洗共同分区的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']]

这里有多个答案,表明加入共分区数据不会导致混乱,这对我来说很有意义.示例:共同分区的RDD的连接是否会导致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并且不会阻止随机播放.

另请参见Spark中的默认分区方案