大数据Spark（七十三）：Transformation转换算子glom和foldByKey使用案例
文章目录
Transformation转换算子glom和foldByKey使用案例
一、glom使用案例
二、foldByKey使用案例
Transformation转换算子glom和foldByKey使用案例
一、glom使用案例
glom可以将每个分区中的数据元素合并为一个数组，将RDD[T]转换成RDD[Array[T]]类型。如果RDD数据量小且需要对分区内的数据进行统计时（最大、最小值统计）可以使用glom。
Java代码:
SparkConf conf = new SparkConf().setMaster("local").setAppName("GlomTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(10, 20, 5, 7, 9, 20, 15, 3, 8), 3);
rdd.mapPartitionsWithIndex(new Function2<Integer, Iterator<Integer>, Iterator<String>>() {@Overridepublic Iterator<String> call(Integer index, Iterator<Integer> iter) throws Exception {ArrayList<String> list = new ArrayList<>();while (iter.hasNext()) {Integer next = iter.next();list.add("rdd1 partition index: " + index + " current value: " + next);}return list.iterator();}
},true).foreach(s->System.out.println(s));//glom: 将RDD每个分区中的元素放到一个集合中，形成RDD
JavaRDD<List<Integer>> glomRDD = rdd.glom();glomRDD.foreach(list->{System.out.println(list);//计算每个分区元素总和int sum = 0;for (Integer i : list) {sum += i;}System.out.println("sum: " + sum);
});sc.stop();
Scala代码：
val conf = new SparkConf().setMaster("local").setAppName("GlomTest")
val sc = new SparkContext(conf)val rdd: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 3)rdd.mapPartitionsWithIndex((index,iter)=>{val list = new ListBuffer[String]()while (iter.hasNext) {list.append(s"rdd partition index: $index ,current value: ${iter.next()}")}list.iterator
},true).foreach(println)val result: RDD[Array[Int]] = rdd.glom()
result.foreach(list=>{println(list.mkString(","))println(list.sum)
})sc.stop()二、foldByKey使用案例
foldByKey针对K,V格式RDD进行数据聚合操作，与reduceByKey类似，该算子是map端有预聚合的算子，但可以为每个分区中的每个不同K提供一个初始值。其函数签名如下：
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]- zeroValue:在map端聚合过程，每个分区中的每个不同K的初始值。
- func:用于在map端和reduce端合并具有相同K的V的函数。
注意：需要对具有相同键的数据进行聚合操作，且分区内和分区间的聚合规则相同时，可以使用foldByKey，如果分区内和分区间的聚合规则不相同可以使用aggregateByKey。
Java代码：
SparkConf conf = new SparkConf().setMaster("local").setAppName("foldByKeyTest");
JavaSparkContext sc = new JavaSparkContext(conf);JavaPairRDD<String, Integer> rdd = sc.parallelizePairs(Arrays.asList(new Tuple2<String, Integer>("a", 1),new Tuple2<String, Integer>("b", 2),new Tuple2<String, Integer>("a", 3),new Tuple2<String, Integer>("b", 4),new Tuple2<String, Integer>("c", 5)
));JavaPairRDD<String, Integer> result = rdd.foldByKey(10, new Function2<Integer, Integer, Integer>() {@Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1 + v2;}
});//结果：a:14 b:16 c:15
result.foreach(tp-> System.out.println(tp));sc.stop();
Scala代码：
val conf = new SparkConf().setMaster("local").setAppName("FoldByKeyTest")
val sc = new SparkContext(conf)val rdd: RDD[(String, Int)] = sc.parallelize(Array(("a", 1),("b", 2),("a", 3),("b", 4),("c", 5)), 3)rdd.foldByKey(10)(_+_).foreach(println)sc.stop()
结果：
#如果分区设置为1个，每个分区不同key初始值为10
结果：a:14 b:16 c:15#如果分区设置为3个，每个分区不同key初始值为10
结果：a:24 b:26 c:15
- 📢博客主页：https://lansonli.blog.csdn.net
- 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
- 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
- 📢停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨