说明
mapPartitionsWithIndex 算子可以得到分区索引
分区索引 与 数组索引类似，从0开始
应用场景
针对某个分区的数据做处理
知道每个数据所在的分区
声明
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false
): RDD[U] = withScope {
val cleanedF = sc.clean(f)
new MapPartitionsRDD(
this,
(_: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(index, iter),
preservesPartitioning)
}
参数：
f: (Int, Iterator[T]) => Iterator[U]：传入函数，该函数有 2 个形参，返回类型是Iterator迭代器。- 第一个形参：
Int类型，分区的索引 - 第二个形参：
Iterator类型，该分区的数据
- 第一个形参：
底层实现： 调用 MapPartitionsRDD
例子
针对第二个分区的数据做处理
//设置2个分区
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val resRDD = rdd.mapPartitionsWithIndex((index,iter)=>{
println("分区索引："+index)
if(index==1){ //对第二个分区的数据做处理
iter.map(_*2) // 乘以2
}else{
Nil.iterator // 返回空迭代器
// Iterator()
}
})
resRDD.collect().foreach(println)
sc.stop()
知道每个数据所在的分区
val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)
//设置2个分区
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
val resRDD = rdd.mapPartitionsWithIndex((index,iter)=>{
println("分区索引："+index)
iter.map({
(index,_) // 新数据格式：(分区索引,数据)
})
})
resRDD.collect().foreach(println)
sc.stop()