阅读 17

Spark 的 cogroup 和 join 算子

cogroup 这个算子使用的频率很低,join 算子使用频率较高,两者都是根据两个 RDD 的 key 进行关联。具体看下面的代码,先看下面的 2 个 RDD:

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

        List<Tuple2<String, Integer>> words1 = Arrays.asList(
                new Tuple2<>("hello", 3),
                new Tuple2<>("hello", 2),
                new Tuple2<>("world", 7),
                new Tuple2<>("hello", 12),
                new Tuple2<>("you", 9)
        );

        List<Tuple2<String, Integer>> words2 = Arrays.asList(
                new Tuple2<>("hello", 21),
                new Tuple2<>("world", 24),
                new Tuple2<>("hello", 25),
                new Tuple2<>("you", 28)
        );

        JavaPairRDD<String, Integer> words1RDD = sc.parallelizePairs(words1);
        JavaPairRDD<String, Integer> words2RDD = sc.parallelizePairs(words2);

复制代码

上面的 RDD 中,words1RDD 和 words2RDD 中的 key 都有重复的。然后看看看两者分别用 cogroup 和 join 算子的操作结果,先看 cogroup 的:

				int count = 1;

        JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> cogroupRDD = words1RDD.cogroup(words2RDD);
        List<Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>>> cogroupResult = cogroupRDD.collect();
        for (Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> t : cogroupResult){
            String word = t._1;
            Iterable<Integer> word1Counts = t._2._1;
            Iterable<Integer> word2Counts = t._2._2;

            String countInfo = "";
            for (Integer c1 : word1Counts) {
                countInfo += c1 + "(words1RDD),";
            }

            for (Integer c2 : word2Counts) {
                countInfo += c2 + "(words2RDD),";
            }

            System.out.println(String.format("第%s个元素为:%s -> %s", count, word, countInfo));

            count++;
        }
复制代码

输出结果为:

1个元素为:you -> 9(words1RDD),28(words2RDD),
第2个元素为:hello -> 3(words1RDD),2(words1RDD),12(words1RDD),21(words2RDD),25(words2RDD),
第3个元素为:world -> 7(words1RDD),24(words2RDD),
复制代码

再看 join 的:

JavaPairRDD<String, Tuple2<Integer, Integer>> joinedRDD = words1RDD.join(words2RDD);
        List<Tuple2<String, Tuple2<Integer, Integer>>> joinedResult = joinedRDD.collect();
        for (Tuple2<String, Tuple2<Integer, Integer>> t : joinedResult) {
            System.out.println(String.format("第%s个元素为:%s -> %s(words1RDD),%s(words2RDD)", count, t._1, t._2._1, t._2._2));
            count++;
        }
复制代码

输出结果为:

1个元素为:you -> 9(words1RDD),28(words2RDD)
第2个元素为:hello -> 3(words1RDD),21(words2RDD)
第3个元素为:hello -> 3(words1RDD),25(words2RDD)
第4个元素为:hello -> 2(words1RDD),21(words2RDD)
第5个元素为:hello -> 2(words1RDD),25(words2RDD)
第6个元素为:hello -> 12(words1RDD),21(words2RDD)
第7个元素为:hello -> 12(words1RDD),25(words2RDD)
第8个元素为:world -> 7(words1RDD),24(words2RDD)
复制代码

cogroup 算子计算过程会对相同的 key 做聚合操作,join 则不会。

关注下面的标签,发现更多相似文章
评论