Spark 的 cogroup 和 join 算子

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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 的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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
2
3
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 的:

1
2
3
4
5
6
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
2
3
4
5
6
7
8
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 则不会。