pyspark - 如何在pySpark中的distnct后压缩

Question

以下程序在 zip 步骤中失败。

x = sc.parallelize([1, 2, 3, 1, 2, 3])
y = sc.parallelize([1, 2, 3])
z = x.distinct()
print x.zip(y).collect()

产生的错误取决于是否指定了多个分区。

我明白那个

两个 RDD [必须] 具有相同数量的分区和每个分区中相同数量的元素。

解决此限制的最佳方法是什么？

我一直在使用以下代码执行操作，但我希望找到更有效的方法。

def safe_zip(left, right):
    ix_left = left.zipWithIndex().map(lambda row: (row[1], row[0]))
    ix_right = right.zipWithIndex().map(lambda row: (row[1], row[0]))
    return ix_left.join(ix_right).sortByKey().values()

Answer 1

我认为这可以通过在您的 RDD 上使用 cartesian() 来完成

import pyspark
x = sc.parallelize([1, 2, 3, 1, 2, 3])
y = sc.parallelize([1, 2, 3])
x.distinct().cartesian(y.distinct()).collect()