如何合并两个rdds

Question

现在我有两个rdds，第一个是这样的

1,2

3,4

5,6

另外一个是这样的

7,8

9,10

11,12

现在我想像这样合并这两个 rdds

1,2,7,8

3,4,9,10

5,6,11,12

我该怎么做？ rdd.union 无法得到这个结果

Answer 1

如果您可以保证两个 RDD 具有相同数量的元素和分区，则可以使用 rdd.zip 获得所需的结果，然后重新塑造结果对：

rdd1.zip(rdd2).map{case ((a1,a2),(b1,b2))=> (a1,a2,b1,b2)}

如果两个 RDD 的元素或分区数量不同，则需要一些键来连接它们。对它们进行索引不是很有效，但可以达到目的，尽管特定领域的解决方案（如果存在）会更好：

val indexed1 = rdd1.zipWithIndex.map(tuple => tuple.swap)
val indexed2 = rdd2.zipWithIndex.map(tuple => tuple.swap)

val joined = indexed1.join(indexed2)
val result = joined.map{case (k,((a1,a2),(b1,b2))) => (a1,a2,b1,b2)}