第17課：RDD案例（join、cogroup等實戰)

本節課通過代碼實戰演示RDD中最重要的兩個算子，join和cogroup

join算子代碼實戰：

//通過代碼演示join算子
val conf = new SparkConf().setAppName("RDDDemo").setMaster("local")
val sc = new SparkContext(conf)
val arr1 = Array(Tuple2(1, "Spark"), Tuple2(2, "Hadoop"), Tuple2(3, "Tachyon"))
val arr2 = Array(Tuple2(1, 100), Tuple2(2, 70), Tuple2(3, 90))
val rdd1 = sc.parallelize(arr1)
val rdd2 = sc.parallelize(arr2)

val rdd3 = rdd1.join(rdd2)
rdd3.collect().foreach(println)

運行結果：

(1,(Spark,100))

(3,(Tachyon,90))

(2,(Hadoop,70))

cogroup算子代碼實戰：

首先通過java的方式編寫：

        SparkConf conf = new SparkConf().setMaster("local").setAppName("Cogroup");

        JavaSparkContext sc = new JavaSparkContext(conf);

        List<Tuple2<Integer, String>> nameList = Arrays.asList(new Tuple2<Integer, String>(1, "Spark"),

                new Tuple2<Integer, String>(2, "Tachyon"), new Tuple2<Integer, String>(3, "Hadoop"));

        List<Tuple2<Integer, Integer>> ScoreList = Arrays.asList(new Tuple2<Integer, Integer>(1, 100),

                new Tuple2<Integer, Integer>(2, 95), new Tuple2<Integer, Integer>(3, 80),

                new Tuple2<Integer, Integer>(1, 80), new Tuple2<Integer, Integer>(2, 110),

                new Tuple2<Integer, Integer>(2, 90));

        JavaPairRDD<Integer, String> names = sc.parallelizePairs(nameList);

        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(ScoreList);

        JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> nameAndScores = names.cogroup(scores);

        nameAndScores.foreach(new VoidFunction<Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>>>() {

            public void call(Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> t) throws Exception {

                System.out.println("ID:" + t._1);

                System.out.println("Name:" + t._2._1);

                System.out.println("Score:" + t._2._2);

            }

        });

       sc.close();

運行結果：

ID:1

Name:[Spark]

Score:[100, 80]

ID:3

Name:[Hadoop]

Score:[80]

ID:2

Name:[Tachyon]

Score:[95, 110, 90]

通過Scala的方式：

val conf = new SparkConf().setAppName("RDDDemo").setMaster("local")
val sc = new SparkContext(conf)
val arr1 = Array(Tuple2(1, "Spark"), Tuple2(2, "Hadoop"), Tuple2(3, "Tachyon"))
val arr2 = Array(Tuple2(1, 100), Tuple2(2, 70), Tuple2(3, 90), Tuple2(1, 95), Tuple2(2, 65), Tuple2(1, 110))
val rdd1 = sc.parallelize(arr1)
val rdd2 = sc.parallelize(arr2)

val rdd3 = rdd1.cogroup(rdd2)
rdd3.collect().foreach(println)
sc.stop()

運行結果：

(1,(CompactBuffer(Spark),CompactBuffer(100, 95, 110)))

(3,(CompactBuffer(Tachyon),CompactBuffer(90)))

(2,(CompactBuffer(Hadoop),CompactBuffer(70, 65)))

備注：

資料來源于：DT_大數據夢工廠（Spark發行版本定制）

更多私密內容，請關注微信公眾號：DT_Spark

如果您對大數據Spark感興趣，可以免費聽由王家林老師每天晚上20：00開設的Spark永久免費公開課，地址YY房間號：68917580