java - 如何有效地连接任意数量的 RDD？

Question

使用 RDD1.join(RDD2) 连接两个 RDD 很简单.但是，如果我在 List<JavaRDD> 中保留任意数量的 RDD , 我怎样才能有效地加入他们？

Answer 1

首先请注意，您不能加入 JavaRDD 。 .您需要获得 JavaPairRDD 通过使用:

• groupBy() (或 keyBy() )
• cartesian()
• [flat]mapToPair()
• zipWithIndex() (很有用，因为它在没有索引的地方添加了索引)
• 等等

然后，一旦你有了你的列表，你就可以像这样加入他们:

JavaPairRDD<Integer, String> linesA = sc.parallelizePairs(Arrays.asList(
                                            new Tuple2<>(1, "a1"),
                                            new Tuple2<>(2, "a2"),
                                            new Tuple2<>(3, "a3"),
                                            new Tuple2<>(4, "a4")));
JavaPairRDD<Integer, String> linesB = sc.parallelizePairs(Arrays.asList(
                                            new Tuple2<>(1, "b1"),
                                            new Tuple2<>(5, "b5"),
                                            new Tuple2<>(3, "b3")));
JavaPairRDD<Integer, String> linesC = sc.parallelizePairs(Arrays.asList(
                                            new Tuple2<>(1, "c1"),
                                            new Tuple2<>(5, "c6"),
                                            new Tuple2<>(6, "c3")));

// the list of RDDs
List<JavaPairRDD<Integer, String>> allLines = Arrays.asList(linesA, linesB, linesC);

// since we probably don't want to modify any of the datasets in the list, we will
// copy the first one in a separate variable to keep the result
JavaPairRDD<Integer, String> res = allLines.get(0);
for (int i = 1; i < allLines.size(); ++i) {  // note we skip position 0 !
    res = res.join(allLines.get(i))
    /*[1]*/  .mapValues(tuple -> tuple._1 + ':' + tuple._2);
}

[1] 行是重要的，因为它映射了一个

JavaPairRDD<Integer, Tuple2<String,String>>回到一个

JavaPairRdd<Integer,String>这使得它与进一步的连接兼容。

根据 chrisw 的回答，这可以像这样放在“一行”中:

JavaPairRDD<Integer, String> res;
res = allLines.stream()
              .reduce((rdd1, rdd2) -> rdd1.join(rdd2).mapValues(tup -> tup._1 + ':' + tup._2))
              .get();  // get value from Optional<JavaPairRDD>

最后，关于性能的一些想法。在上面的示例中，我使用字符串连接将连接的结果缩减回相同类型的 RDD。如果你有很多 RDD，你可能会通过使用 for loop 来加快速度。带有 JavaPairRDD<Integer, StringBuilder> res 的版本，在那里你第一次手工加入。如果需要，我会发布更多详细信息。