PySpark Join改组了共同分区的RDD

Def_Os 发表于 Dev

Def_Os

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']]

在SO上有多个答案，这表明加入共分区数据不会引起洗牌，这对我来说很有意义。示例：联合分区的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。

零323

在这种情况下，既不分区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中的默认分区方案

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。