算子_综合_代码迷

MapPartitionsRDD 算子的程序与应用

程序底层执行流程,

1.连接Spark配置文件,设置项目名称,和创建本地地址,

2.创建连接,创建RDD,分三个去,把9个数字均匀的分配到三个去里面,调用RDD,对RDD操作,本质上是对RDD里面的每一个分区进行操作,每一个分区有对应一个Task,Task里面的逻辑也就是分区对应的计算逻辑,

3.调用map,从下面的程序中可以得知,每一条数据都会调用一次map函数,这样就降低数据处理的效率,

4.分析一下map底层的实现逻辑,在调用map方法的时候,先看这个函数是否可以序列化,在new MapPartitionsRDD[] ,把上次的partition传进去,也就是新的RDD持有老的RDD的引用,并传入新的迭代器里面,在mapPartitoinRDD里面传的是,上下文,分区编号,迭代器(TaskContext,Int,Iterator) ,通过迭代器实现对里面数据的处理,

5.同时里面有一个方法叫做computer,computer会应用传进来的函数,分别传进上下文,分区编号,因为要对数据进行运算, // 所以要到上一个迭代器拿数据,所以传进去一个父类的迭代器 // (context,split.index,firstParent[T].iterator(split,context)) ,而这个iterator看缓存是否存在数据, // 如果缓存有数据,那么直接到缓存里面直接取,如果没缓存.那么直接从源头拿过来

  def main(args: Array[String]): Unit = {//上传连接机器的名字,和主机地址为本地地址val conf = new SparkConf().setAppName("MapPartitionsDemo").setMaster("local[*]")val sc = new SparkContext(conf)//创建RDD,并分三个分区//会把9个数据均匀的分配到中三个区域里面val words = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)//调用RDD,对RDD进行操作本质上是对RDD里面的每一个分区进行操作,一个分区对应一个Task,Task里面的逻辑//也就是对分区进行计算的逻辑//调用map方法//val re = words.map(_ * 100)val result = words.map(x => {  //这种模式就要一条数据调用一次map,相对来说效率比较低//可以直接取TaskContext,然后调用partition的Idval index = TaskContext.getPartitionId()(index,x*100)})//写一个相对路径result.saveAsTextFile("map-out")sc.stop()//调用map方法本质上是看这个函数是不是可以序列化的,在new MapPartitionsRDD[],把上一次的partition//传进去,也就是新的RDD持有老的RDD得引用,在新的迭代器里面,//在mapPartitionsRDD里面传进去的是上下文,分区编号,迭代器(TaskContext,Int,Iterator),对立面的数据进行处理//输出一个新的迭代器Iterator,(当Task在executor里面被执行的时候会产生新的iterator)//同时里面有一个方法叫做computer,computer会应用传进来的函数,分别传进上下文,分区编号,因为要对数据进行运算,// 所以要到上一个迭代器拿数据,所以传进去一个父类的迭代器// (context,split.index,firstParent[T].iterator(split,context)) ,而这个iterator看缓存是否存在数据,// 如果缓存有数据,那么直接到缓存里面直接取,如果没缓存.那么直接从源头拿过来}

第二段代码,是通过调用mapPartitions,实现将数据以分区单位的方式取出来,一个分区就是一个迭代器,然后再迭代器里面通过调用map方法对数据进行处理,这样

底层代码实现

1.查看程序是否可执行,

2.创建一个新的迭代器,传入TaskContext,Int,Iterator,这不是对迭代器进行map操作,而是把把所写的函数写到迭代器limian,

然后传到新的迭代器里面,所以后面传入的是一个分区的迭代器,这样就可以操作一个分区的数据了(实在不懂可以多看源码)

 //调用mapPatitions方法,可将数据以分区单位取出来,一个分区就是一个迭代器//底层代码,先new一个新的MapPartitionsRDD,里面传,上下文,分区编号,应用传入的参数放进迭代器里面,// 也就是传入的函数得到的是一个迭代器,//it的迭代器,对mapPartitons的要求是输入的是迭代器,返回的也是迭代器val result2 = words.mapPartitions(it=>{//这个是得到每个分区的分区编号val index = TaskContext.getPartitionId()   //这个以后有几个分区就调用几次,调用一个分区后,//开始执行分区里面的相关逻辑
//调用map,这个map是迭代器里面的方法,返回一个新的迭代器,第一个int是分区编号,
val Iterator1: Iterator[(Int, Int)] = it.map(x => { //注意x是局部变量,输入一个x  输出x*100
(index, x * 100)
})
Iterator1})result2.saveAsTextFile("map-out02")sc.stop()}

这个也是对先分区,然后再对个区里面的内容进行操作,有多少分区调用几次map

//输入的是一个迭代器,返回的也是一个迭代器val result4 = words.mapPartitionsWithIndex((index: Int, it: Iterator[Int]) => {it.map(e => {s"partitionIndex:$index,element:$e"})})/*val r = result4.collect()println(r)sc.stop()*/result4.saveAsTextFile("map-out03")sc.stop()

mapPartition和mapPartitionsWithIndex在什么情况下使用,都可以实现map方法,filter方法,但最终要返回新的map迭代器,而mapPartitionsWithIndex,在底层比mapPartition多传入了一个index,