说明
将待处理的数据以 分区为单位 发送到计算节点进行 任意处理
注意： 将整个分区的数据加载到内存，处理数据后，是不会释放的。
优点
速度快，将数据加载到内存
缺点
在内存较小、数据量较大时，容易引发内存溢出
声明
def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]
参数：
- Iterator：是迭代器，返回也是一个迭代器
例子
让每个分区的元素乘以2
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val resRDD = rdd.mapPartitions(iter=>{
println("===")
// iter.filter(_==2)
iter.map(_*2)
})
resRDD.collect().foreach(println)
sc.stop()
找出2的元素
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val resRDD = rdd.mapPartitions(iter=>{
println("===")
iter.filter(_==2)
})
resRDD.collect().foreach(println)
sc.stop()
找出每个分区中最大的元素
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val resRDD = rdd.mapPartitions(iter=>{
println("===")
List(iter.max).iterator //要求返回迭代器，所以要转换
})
与 map()的区别
代码区别
map 方式：
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
//分区数设置为1
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),1)
val resRDD = rdd.map(e=> {
println("----")//有4个元素，会打印4次，说明此代码执行4次
e * 2
}
)
resRDD.collect()
resRDD.foreach(println)
sc.stop()
mapPartitions方式：
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
//分区数设置为1
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),1)
//传入的是迭代器
val resRDD = rdd.mapPartitions(iter=>{
println("===")//打印1次，说明1个分区只执行1次
it.map( e => {
println("!!!!") //打印多次，通过迭代器循环
e * 2
})
})
resRDD.collect().foreach(println)
sc.stop()
一次接收所有的partition数据。只要执行一次就可以了，性能比较高。
速度比 map() 快
数据处理不同
map 算子是分区内一个数据一个数据的执行，类似于串行操作。
mapPartitions 算子是以分区为单位进行批处理操作。
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6,7,8),2)
//以分区为单位进行批处理操作
val mp: RDD[Int] = rdd.mapPartitions(it => {
println("========"+it.mkString(","))
it.map(_ * 2)
})
mp.foreach(println)
sc.stop()
}
执行结果：
========5,6,7,8
========1,2,3,4
说明将分区内的数据一次性传入
功能不同
map 算子主要目的将数据源中的数据进行 转换 和 改变。但是 不会增加、减少 数据。
MapPartitions 算子需要传递一个迭代器，返回一个迭代器，可以 增加 、减少 数据
val rdd2 = rdd.mapPartitions(it => {
println("----===")
it.drop(2) // 删除前2个元素
Array(1,20,300,400,500,-20).iterator //返回另一个迭代器
})
性能的角度
map 算子因为类似于 串行 操作，所以 性能比较低，而是 mapPartitions 算子类似于批处理，所以性能较高。
mapPartitions 算子会 长时间占用内存，那么这样会导致内存可能不够用，出现 内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map 操作。