三.Spark高级算子

1、mapPartitionsWithIndex

把每个partition中的分区号和对应的值拿出来

接收一个函数参数:

第一个参数:分区号

第二个参数:分区中的元素

示例:将每个分区中的元素和分区号打印出来。

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)

创建一个函数返回RDD中的每个分区号和元素:

def func1(index:Int, iter:Iterator[Int]):Iterator[String] ={

iter.toList.map( x => "[PartID:" + index + ", value=" + x + "]" ).iterator}

调用:rdd1.mapPartitionsWithIndex(func1).collect

2、aggregate

先对局部聚合,再对全局聚合

示例:val rdd1 = sc.parallelize(List(1,2,3,4,5), 2)

查看每个分区中的元素:

将每个分区中的最大值求和,注意:初始值是0;

如果初始值时候10,则结果为:30

如果是求和,注意:初始值是0:

如果初始值是10,则结果是:45

一个字符串的例子:

val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)

修改一下刚才的查看分区元素的函数

def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {

iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator

}

两个分区中的元素:

[partID:0, val: a], [partID:0, val: b], [partID:0, val: c],

[partID:1, val: d], [partID:1, val: e], [partID:1, val: f]

运行结果:

更复杂一点的例子

val rdd3 = sc.parallelize(List("12","23","345","4567"),2)

rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x +

y)

结果可能是:”24”,也可能是:”42”

val rdd4 = sc.parallelize(List("12","23","345",""),2)

rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x +

y)

结果是:”10”,也可能是”01”,

原因:注意有个初始值””,其长度0,然后0.toString变成字符串

val rdd5 = sc.parallelize(List("12","23","","345"),2)

rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x +

y)

结果是:”11”,原因同上。

3、aggregateByKey

准备数据:

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat",

12), ("dog", 12), ("mouse", 2)), 2)

def func3(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {

iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator

}

两个分区中的元素:

示例:

将每个分区中的动物最多的个数求和

scala> pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect

res69: Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))

将每种动物个数求和

scala> pairRDD.aggregateByKey(0)(_+_, _ + _).collect

res71: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))

这个例子也可以使用:reduceByKey

scala> pairRDD.reduceByKey(_+_).collect

res73: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))

4、coalesce与repartition

都是将RDD中的分区进行重分区。

区别是:coalesce默认不会进行shufflfflffle(false);而repartition会进行shufflfflffle(true),即:会将数据真正通过网络进行重分区。

示例:

def func4(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {

iter.toList.map(x => "[partID:" + index + ", val: " + x + "]").iterator

}

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)

下面两句话是等价的:

val rdd2 = rdd1.repartition(3)

val rdd3 = rdd1.coalesce(3,true) --->如果是false,查看RDD的length依然是2

5、其他高级算子

http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容