首先看一段代码
public static void cogroup2() {
// 创建SparkConf
SparkConf sparkConf = new SparkConf().setAppName("cogroupJava").setMaster("local");
// 创建JavaSparkContext
JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
// 创建集合
List<Tuple2<String, Integer>> words1 = Arrays.asList(
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("world", 1),
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("you", 1)
);
List<Tuple2<String, Integer>> words2 = Arrays.asList(
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("world", 1),
new Tuple2<String, Integer>("hello", 1),
new Tuple2<String, Integer>("you", 1)
);
// 并行化集合，创建初始化RDD
JavaPairRDD<String, Integer> words1RDD = javaSparkContext.parallelizePairs(words1);
JavaPairRDD<String, Integer> words2RDD = javaSparkContext.parallelizePairs(words2);
// 使用cogroup算子关联两个RDD
// 相当于是，一个key join上的所有value，都给放到一个Iterable里面去了
// cogroup，不太好讲解，希望通过动手编写我们的案例，仔细体会其中的奥妙
JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> studentScore = words1RDD.cogroup(words2RDD);
studentScore.foreach(new VoidFunction<Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>>>() {
@Override
public void call(Tuple2<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> t) throws Exception {
System.out.println(t._1);
System.out.println(t._2._1);
System.out.println(t._2._2);
System.out.println("===============================");
}
});
// 关闭javaSparkContext
javaSparkContext.close();
}
看图
cogroup.png
cogroup算子
- 基础的算子
- 在我们大量的实践中，很少遇到说要用cogroup算子的情况
- cogroup算子是其他很多算子的基础，比如join
可以把上面那段代码跑一下，其实就是将两个个rdd的key对应的value值分别封装到一个Iterator中去
