spark2.x由淺入深深到底系列六之RDD java api詳解四

學(xué)習(xí)spark任何的知識(shí)點(diǎn)之前，先對(duì)spark要有一個(gè)正確的理解，可以參考：正確理解spark

本文對(duì)join相關(guān)的api做了一個(gè)解釋

SparkConf conf = new SparkConf().setAppName("appName").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);

JavaPairRDD<Integer, Integer> javaPairRDD =
        sc.parallelizePairs(Arrays.asList(new Tuple2<>(1, 2),
                new Tuple2<>(3, 4), new Tuple2<>(3, 6), new Tuple2<>(5, 6)));
JavaPairRDD<Integer, Integer> otherJavaPairRDD =
        sc.parallelizePairs(Arrays.asList(new Tuple2<>(3, 9),
                new Tuple2<>(4, 5)));
//結(jié)果： [(4,([],[5])), (1,([2],[])), (3,([4, 6],[9])), (5,([6],[]))]
System.out.println(javaPairRDD.cogroup(otherJavaPairRDD).collect());

//結(jié)果： [(4,([],[5])), (1,([2],[])), (3,([4, 6],[9])), (5,([6],[]))]
// groupWith和cogroup效果是一模一樣的
System.out.println(javaPairRDD.groupWith(otherJavaPairRDD).collect());

//結(jié)果： [(3,(4,9)), (3,(6,9))]
//基于cogroup實(shí)現(xiàn)的，就是取cogroup結(jié)果中相同key在兩個(gè)RDD都有value的數(shù)據(jù)
System.out.println(javaPairRDD.join(otherJavaPairRDD).collect());

//結(jié)果： [(1,(2,Optional.empty)), (3,(4,Optional[9])), (3,(6,Optional[9])), (5,(6,Optional.empty))]
//基于cogroup實(shí)現(xiàn)的，結(jié)果需要出現(xiàn)的key以左邊的RDD為準(zhǔn)
System.out.println(javaPairRDD.leftOuterJoin(otherJavaPairRDD).collect());

//結(jié)果： [(4,(Optional.empty,5)), (3,(Optional[4],9)), (3,(Optional[6],9))]
//基于cogroup實(shí)現(xiàn)的，結(jié)果需要出現(xiàn)的key以右邊的RDD為準(zhǔn)
System.out.println(javaPairRDD.rightOuterJoin(otherJavaPairRDD).collect());

//結(jié)果： [(4,(Optional.empty,Optional[5])), (1,(Optional[2],Optional.empty)), (3,(Optional[4],Optional[9])), (3,(Optional[6],Optional[9])), (5,(Optional[6],Optional.empty))]
//基于cogroup實(shí)現(xiàn)的，結(jié)果需要出現(xiàn)的key是兩個(gè)RDD中所有的key
System.out.println(javaPairRDD.fullOuterJoin(otherJavaPairRDD).collect());

從上可以看出，最基本的操作是cogroup這個(gè)操作，下面是cougroup的原理圖：

如果想對(duì)cogroup原理更徹底的理解，可以參考：spark core RDD api原理詳解