文章目录

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>>() {
    @Override
    public 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>() {
    @Override
    public 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博客🙉
• 📢停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨