当前位置：首页 > news >正文

大数据Spark（八十）：Action行动算子fold和aggregate使用案例

news 2026/3/27 0:04:02

文章目录

Action行动算子fold和aggregate使用案例

一、fold使用案例

二、aggregate使用案例

Action行动算子fold和aggregate使用案例

一、fold使用案例

fold用于对RDD中的元素进行聚合操作，最终返回一个结果。类似reduce算子，但与reduce不同的是其可以对每个分区中的数据提供一个初始值，让分区中的数据与该初始值进行聚合，最终该初始值还会与各个分区的结果再次聚合。

fold的函数签名如下：

def fold(zeroValue: T)(op: (T, T) => T): T

zeroValue：聚合操作的初始值，类型为 T。
op：用于合并元素的二元操作函数。

fold的工作原理：在每个分区内，fold 使用初始值 zeroValue 和二元操作函数 op，将该分区内的所有元素进行聚合。在所有分区内的聚合完成后，fold 将各分区的结果与初始值 zeroValue 一起，使用相同的二元操作函数 op 进行全局聚合，得到最终结果。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("FoldTest"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a","b","c","d","e","f"), 3); rdd.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() { @Override public Iterator<String> call(Integer index, Iterator<String> iter) throws Exception { ArrayList<String> list = new ArrayList<>(); while (iter.hasNext()) { String next = iter.next(); list.add("rdd partition index: " + index + " current value: " + next); } return list.iterator(); } },true).foreach(x-> System.out.println(x)); /** * 0号分区：a b * 1号分区：c d * 2号分区：e f * * 0号分区:hello~a~b * 1号分区:hello~c~d * 2号分区:hello~e~f * * 最终结果：hello~hello~a~b~hello~c~d~hello~e~f */ String str = rdd.fold("hello", new Function2<String, String, String>() { @Override public String call(String v1, String v2) throws Exception { return v1 + "~" + v2; } }); System.out.println(str); sc.stop();

Scala代码：

val conf = new SparkConf().setMaster("local").setAppName("FoldTest") val sc = new SparkContext(conf) val rdd: RDD[String] = sc.parallelize(List("a", "b", "c", "d", "e", "f"), 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 }).foreach(println) /** * 0号分区：a b * 1号分区：c d * 2号分区：e f * map端聚合： * 0号分区:hello~a~b * 1号分区:hello~c~d * 2号分区:hello~e~f * * 最终结果：hello~hello~a~b~hello~c~d~hello~e~f */ val result: String = rdd.fold("hello")((v1, v2) => { v1 + "~" + v2 }) println(result) sc.stop()

二、aggregate使用案例

aggregate用于对RDD中的元素进行聚合操作，最终返回一个结果。与 fold 和 reduce 等算子不同，aggregate 允许用户分别定义分区内和分区间的聚合函数，提供了更大的灵活性。

aggregate函数签名如下：

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

zeroValue：聚合操作的初始值，类型为 U。
seqOp：分区内的聚合函数，用于将分区内的元素与累加器进行合并，类型为 (U, T) => U。
combOp：分区间的聚合函数，用于将不同分区的累加器结果进行合并，类型为 (U, U) => U

aggregate工作原理：在每个分区内，使用初始值 zeroValue 和函数 seqOp，将该分区内的所有元素进行聚合。在所有分区内的聚合完成后，使用初始值 zeroValue 和函数 combOp，将各分区的结果进行全局聚合，得到最终结果。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("AggregateTest"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a","b","c","d","e","f"), 3); rdd.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() { @Override public Iterator<String> call(Integer index, Iterator<String> iter) throws Exception { ArrayList<String> list = new ArrayList<>(); while (iter.hasNext()) { String next = iter.next(); list.add("rdd partition index: " + index + " current value: " + next); } return list.iterator(); } },true).foreach(x-> System.out.println(x)); /** * 0号分区：a b * 1号分区：c d * 2号分区：e f * * map端聚合： * 0号分区:hello~a~b * 1号分区:hello~c~d * 2号分区:hello~e~f * * 最终结果：hello@hello~a~b@hello~c~d@hello~e~f */ String result = rdd.aggregate("hello", new Function2<String, String, String>() { @Override public String call(String s1, String s2) throws Exception { return s1 + "~" + s2; } }, new Function2<String, String, String>() { @Override public String call(String s1, String s2) throws Exception { return s1 + "@" + s2; } }); System.out.println(result); sc.stop();

Scala代码：

val conf = new SparkConf().setMaster("local").setAppName("FoldTest") val sc = new SparkContext(conf) val rdd: RDD[String] = sc.parallelize(List("a", "b", "c", "d", "e", "f"), 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 }).foreach(println) val result: String = rdd.aggregate("hello")( (v1, v2) => { v1 + "~" + v2 }, (v1, v2) => { v1 + "@" + v2 } ) println(result) sc.stop()