spark rdd 转换过程

 

从一个简单的例子，来看rdd的演化，和stage是如何生成的（过程灰常之复杂和抽象，请参考附件的图来理解） 

object BaiWordCount2 {
  def main(args: Array[String]) {
    .....
    // Create the context
    val ssc = new SparkContext(args(0), "BaiWordCount", System.getenv("SPARK_HOME"), Seq(System.getenv("SPARK_EXAMPLES_JAR")))
    val lines = ssc.textFile(args(1))//基于hadoopRdd，创建了一个MapRdd[String]
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x =>{ println("x:"+ x);(x, 1) } )//这回返回的是一个元组了
    val red = wordCounts.reduceByKey( (a,b)=>{a + b} )
    red.saveAsTextFile("/root/Desktop/out")
  }
}

我们首先总结出每个操作生成的rdd，这是通过源码追踪得到的： 
textFile：生成一个HadoopRDD 和 一个MapPedRDD[String] 
flatMap：前面生成的MapPedRDD[String]，调用flatMap函数，生成了FlatMaPPedRDD[String] 
map:FlatMaPPedRDD[String],调用map函数，创建了一个创建了一个MapPartitionsRDD，再通过隐式转换，生成了PairRDD[String,Int]（因为map操作，产生了一对值key和value） 
reduceByKey:生成三个Rdd，首先根据之前的PairRDD，生成一个MapPartitionsRDD（这个RDD起到类似map-reduce里面，combine（）的作用），再生成一个shuffledRdd（这个rdd是分割stage的重要依据之一），这之后再生成一个MapPartitionsRDD[String](这个rdd起到hadoop里reducer的作用) 
saveAsTextFile先生成一个MapPedRDD，然后调用runJob函数，将之前生成的rdd链表，提交到spark集群上，spark根局rdd的类型，划分成一个或多个stage（只有shuffledRdd这类的rdd，才会成为stage和stage之间的边界），然后将各个stage，按照依赖的先后顺序，将stage先后提交集群进行计算 
下边通过textFile来详细说明rdd链表的生成过程和主要数据结构，主要注意deps和几种dependency： 
从rdd生成的方式来说可以分成四类：通过外部数据生成rdd，通过transformations函数生成，缓存操作，actions操作 
textFile函数无疑属于第一种，通过外部数据，生成rdd，输入的文件路径，可以是hdfs://开头的hdfs数据，也可以本地文件路径，例如"/root/Desktop/word.text" 
textFile函数调用hadoopFile 函数，生成一个hadoopRdd[K,V],默认情况下，泛型参数K和V，对应HadoopRDD的构造函数里的keyClass和valueClass。 
也就是一个Rdd[LongWritable,Text],通过外部数据生成rdd的第一个rdd的特点是，deps是一个空的list，原因是它是从外部文件生成的，没有父rdd。 
生成了Rdd[LongWritable,Text]后，还要调用transformations函数map：map(pair => pair._2.toString)，来生成一个MappedRDD 
MappedRDD(this, sc.clean(f))，这里this，就是之前生成的HadoopRDD，MappedRDD的构造函数，会调用父类的构造函数RDD[U](prev)， 
这个this(也就是hadoopRdd),会被赋值给prev，然后调用RDD.scala中，下面的构造函数 

def this(@transient oneParent: RDD[_]) =
  this(oneParent.context , List(new OneToOneDependency(oneParent)))

这个函数的作用，是把父RDD的SparkContext(oneParent.context),和一个列表List(new OneToOneDependency(oneParent)))，传入了另一个RDD的构造函数， 

RDD[T: ClassManifest](
   @transient private var sc: SparkContext,
   @transient private var deps: Seq[Dependency[_]]
 )

这样我们可以看到，在所有有父子关系的RDD，共享的是同一个SparkContext。而子RDD的deps变量，也被赋值为一个List，里面包含一个OneToOneDependency实例，表明父RDD和子RDD之间的关系 
其实大多数的父子关系，包含的都是OneToOneDependency.比较例外的几个,比如join，这个很明显，他的数据不是来自同一个父RDD。而shuffledRdd的Dependency是ShuffledDependency 

父Rdd会在子rdd的构造函数中被传入，然后放入子rdd实例的deps里面，被记录下来。这样，当我们得到一个Rdd之后，就可以向后回溯它的祖先，再结合传入的函数变量f，完整的得到它的构造过程。 
flatMap，map，reduceByKey，saveAsTextFile则按顺序创建各自的rdd，然后在deps中记录父rdd，同时根据rdd的类型，生成各自的不同类型的dependency。 
在saveAsTextFile函数把整个计算任务提交到集群之前，所有的函数进行的操作，仅仅就是生成rdd链表而已。saveAsTextFile是action类型的操作，action的共同特点是，会调用RunJob一类的函数，调用Dagscheduler.runJob,将最后一个rdd（在我们这个例子里，就是saveAsTextFile生成的那个MappedRdd），提交到集群上。集群会以这个rdd为参数之一，生成一个stage，名叫finalStage（故名思意，这是最终的一个stage）。然后调用submitStage，将刚刚生成的finalStage提交到集群上。这个stage是否会被马上执行呢？不一定，因为程序会调用getMissingParentStages，进行寻找，是否有需要先进行提交的stage---这个过程可以这样类比，一个查询操作，在提交之后，要先检查是否有子查询，如果有，先执行子查询，然后在执行父查询，这里的原因很简单，父查询依赖于子查询的数据。同理，在stage执行的过程中，也要先查询，它是否需要其他stage的数据（其实之后一种数据，就是通过shuffle传过来的数据），如果有，那么这些stage，就是它的missingParentStage，它要等他的missingParentStage执行成功，然后通过shuffle机制把数据传给它，才能开始执行。这个过程的执行过程如下：从最后一个rdd起，查看它的dependency的类型，如果是 
shuffledDependency，则创建一个ShuffleMapStage,否则，就向前遍历，依次递归，知道最前面的rdd为止 

private def getMissingParentStages(stage: Stage): List[Stage] = {
   .....
   def visit(rdd: RDD[_]) {
     if (!visited(rdd)) {/
       visited += rdd
       if (getCacheLocs(rdd).contains(Nil)) {
         for (dep <- rdd.dependencies) {
           dep match {
             case shufDep: ShuffleDependency[_,_] =>
              val mapStage = getShuffleMapStage(shufDep, stage.jobId)
             if (!mapStage.isAvailable) {
                 missing += mapStage
               }
             case narrowDep: NarrowDependency[_] =>
               visit(narrowDep.rdd)
           }
         }
       }
     }
   }
   visit(stage.rdd)
   missing.toList
 }

当getMissingParentStage（stage）的结果为空的时候，表明这个stage没有missingParentStage，或者它的missingParentStage已经都执行完了，则当前这个stage才能被成功的提交到集群去执行，否则，它就要等待，并重复调用getMissingParentStage，直到它的结果为空，才可以被提交。 

 

 

http://baishuo491.iteye.com/blog/2019510