标签：连接 apach map partition art array 比较 元素 初始化
mapPartitions
与map类似，map函数是应用到每个元素，而mapPartitions的输入函数是每个分区的数据，把每个分区中的内容作为整体来处理的。 当map里面有比较耗时的初始化操作时，比如连接db,可以采用mapPartitions,它对每个partition操作一次，其函数的输入与输出都是iterator类型。
实例如下：
scala> val rdd1=sc.parallelize(1 to 9,3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> def myfunc[T](iter:Iterator[T]):Iterator[(T,T)]={
| var res=List[(T,T)]()
| var pre=iter.next
| while (iter.hasNext) {
| val cur=iter.next
| res.::=(pre,cur)
| pre=cur
| }
| res.iterator
| }
myfunc: [T](iter: Iterator[T])Iterator[(T, T)]
scala> rdd1.mapPartitions(myfunc)
res2: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[2] at mapPartitions at <console>:28
scala> res2.collect()
res3: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))
mapPartitionsWithIndex
与 mapPartitions 类似，参数需多传一个分区的index.
实例如下：
scala> val mapReslut=rdd1.mapPartitionsWithIndex{
| (index,iterator)=>{
| val list=iterator.toList
| list.map(x=>x +"->"+index).iterator
| }
| }
mapReslut: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at mapPartitionsWithIndex at <console>:25
scala> mapReslut.collect
res6: Array[String] = Array(1->0, 2->0, 3->0, 4->1, 5->1, 6->1, 7->2, 8->2, 9->2)
Spark mapPartitions 及mapPartitionsWithIndex算子
标签：连接 apach map partition art array 比较 元素 初始化
原文地址：https://www.cnblogs.com/abcdwxc/p/9865071.html
