任意の数のRDDに効率的に参加するにはどうすればいいですか？

Question

RDD1.join(RDD2)で2つのRDDに簡単に結合できます。しかし、List<JavaRDD>に任意の数のRDDを保持すると、どうすれば効率的にそれらに参加できますか？

Answer 1

最初に、​​に参加できないことに注意してください。 （どれもがない場合、それは、インデックスを追加するために有用）など

[flat]mapToPair()

zipWithIndex()

cartesian()

• groupBy()（またはkeyBy()）を
：あなたが使用して JavaPairRDDを取得する必要があります

次に、 ISTは、あなたは、このようにそれらすべてに参加することができます。それは、さらに加入との互換性になり

JavaPairRdd<Integer,String>に戻す

JavaPairRDD<Integer, Tuple2<String,String>>をマップするため

が

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]とラインは、重要なものです。

chrisw回答に基づいて、これはこのような「1行」に入れることができます：パフォーマンス上の

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バージョンを使用して、これを少し早めることができます。ここでは最初に手で参加します。必要に応じて詳細を掲載します。

Answer 2

私はJavaRDDクラス/インタフェースに慣れていないんだけど、おそらくあなたは、Stringクラスでhttps://docs.oracle.com/javase/tutorial/collections/streams/reduction.html

final List<JavaRDD> list = getList(); // where getList is your list implementation containing JavaRDD instances 

// The JavaRDD class provides rdd() to get the RDD 
final JavaRDD rdd = list.stream().map(JavaRDD::rdd).reduce(RDD::join); 

例を参照して、Javaの8に高階関数reduceを使用してこの問題を解決できるだろう何かのように： -

を生成

Stream.of("foo", "bar", "baz").reduce(String::concat); 

foob​​arbaz