准备
1. spark已经安装完毕
2. spark运行在local mode或local-cluster
mode
local-cluster mode
local-cluster模式也称为伪分布式，可以使用如下指令运行
MASTER=local[1,2,1024] bin/spark-shell[1,2,1024] 分别表示，executor number, core number和内存大小，其中内存大小不应小于默认的512M
Driver Programme的初始化过程分析
初始化过程的涉及的主要源文件
1. SparkContext.scala 整个初始化过程的入口
2. SparkEnv.scala 创建BlockManager,MapOutputTrackerMaster, ConnectionManager, CacheManager
3. DAGScheduler.scala 任务提交的入口，即将Job划分成各个stage的关键
4. TaskSchedulerImpl.scala决定每个stage可以运行几个task，每个task分别在哪个executor上运行
5. SchedulerBackend
1. 最简单的单机运行模式的话，看LocalBackend.scala
2. 如果是集群模式，看源文件SparkDeploySchedulerBackend
初始化过程步骤详解
步骤1：根据初始化入参生成SparkConf，再根据SparkConf来创建SparkEnv,
SparkEnv中主要包含以下关键性组件 1. BlockManager 2. MapOutputTracker 3. ShuffleFetcher 4.ConnectionManagerprivate[spark] val env = SparkEnv.create(conf, "", conf.get("spark.driver.host"), conf.get("spark.driver.port").toInt, isDriver = true, isLocal = isLocal) SparkEnv.set(env)步骤2：创建TaskScheduler,根据Spark的运行模式来选择相应的SchedulerBackend，同时启动taskscheduler，这一步至为关键
private[spark] var taskScheduler = SparkContext.createTaskScheduler(this, master, appName) taskScheduler.start()TaskScheduler.start目的是启动相应的SchedulerBackend，并启动定时器进行检测
overridedef start() { backend.start() if (!isLocal && conf.getBoolean("spark.speculation", false)) { logInfo("Starting speculative execution thread") import sc.env.actorSystem.dispatcher sc.env.actorSystem.scheduler.schedule(SPECULATION_INTERVAL milliseconds, SPECULATION_INTERVAL milliseconds) { checkSpeculatableTasks() } } }步骤3：以上一步中创建的TaskScheduler实例为入参创建DAGScheduler并启动运行
@volatileprivate[spark] var dagScheduler = new DAGScheduler(taskScheduler) dagScheduler.start()步骤4：启动WEB UI
ui.start()RDD的转换过程
还是以最简单的wordcount为例说明rdd的转换过程
sc.textFile("README.md").flatMap(line=>line.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)上述一行简短的代码其实发生了很复杂的RDD转换，下面仔细解释每一步的转换过程和转换结果
步骤1：val rawFile = sc.textFile("README.md")
textFile先是生成hadoopRDD，然后再通过map操作生成MappedRDD,如果在spark-shell中执行上述语句，得到的结果可以证明所做的分析
scala> sc.textFile("README.md") 14/04/2313:11:48 WARN SizeEstimator: Failed to check whether UseCompressedOops is set; assuming yes 14/04/2313:11:48 INFO MemoryStore: ensureFreeSpace(119741) called with curMem=0, maxMem=311387750 14/04/2313:11:48 INFO MemoryStore: Block broadcast_0 stored as values to memory (estimated size 116.9 KB, free 296.8 MB) 14/04/2313:11:48 DEBUG BlockManager: Put block broadcast_0 locally took 277 ms 14/04/2313:11:48 DEBUG BlockManager: Put for block broadcast_0 without replication took 281 ms res0: org.apache.spark.rdd.RDD[String] = MappedRDD[1] at textFile at :13步骤2: val splittedText = rawFile.flatMap(line => line.split(""))
flatMap将原来的MappedRDD转换成为FlatMappedRDD
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] = new FlatMappedRDD(this, sc.clean(f))步骤3：val wordCount = splittedText.map(word => (word, 1))
利用word生成相应的键值对，上一步的FlatMappedRDD被转换成为MappedRDD
步骤4：val reduceJob = wordCount.reduceByKey(_ + _)，这一步最复杂
步骤2，3中使用到的operation全部定义在RDD.scala中，而这里使用到的reduceByKey却在RDD.scala中见不到踪迹。reduceByKey的定义出现在源文件PairRDDFunctions.scala
细心的你一定会问reduceByKey不是MappedRDD的属性和方法啊，怎么能被MappedRDD调用呢？其实这背后发生了一个隐式的转换，该转换将MappedRDD转换成为PairRDDFunctions
implicit def rddToPairRDDFunctions[K: ClassTag, V: ClassTag](rdd: RDD[(K, V)]) = new PairRDDFunctions(rdd)这种隐式的转换是scala的一个语法特征，如果想知道的更多，请用关键字"scalaimplicit method"进行查询，会有不少的文章对此进行详尽的介绍。
接下来再看一看reduceByKey的定义
def reduceByKey(func: (V, V) => V): RDD[(K, V)] = { reduceByKey(defaultPartitioner(self), func) } def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = { combineByKey[V]((v: V) => v, func, func, partitioner) } def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializerClass: String = null): RDD[(K, C)] = { if (getKeyClass().isArray) { if (mapSideCombine) { thrownew SparkException("Cannot use map-side combining with array keys.") } if (partitioner.isInstanceOf[HashPartitioner]) { thrownew SparkException("Default partitioner cannot partition array keys.") } } val aggregator = new Aggregator[K, V, C](createCombiner, mergeValue, mergeCombiners) if (self.partitioner == Some(partitioner)) { self.mapPartitionsWithContext((context, iter) => { new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, context)) }, preservesPartitioning = true) } elseif (mapSideCombine) { val combined = self.mapPartitionsWithContext((context, iter) => { aggregator.combineValuesByKey(iter, context) }, preservesPartitioning = true) val partitioned = new ShuffledRDD[K, C, (K, C)](combined, partitioner) .setSerializer(serializerClass) partitioned.mapPartitionsWithContext((context, iter) => { new InterruptibleIterator(context, aggregator.combineCombinersByKey(iter, context)) }, preservesPartitioning = true) } else { // Don't apply map-side combiner. val values = new ShuffledRDD[K, V, (K, V)](self, partitioner).setSerializer(serializerClass) values.mapPartitionsWithContext((context, iter) => { new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, context)) }, preservesPartitioning = true) } }reduceByKey最终会调用combineByKey,在这个函数中PairedRDDFunctions会被转换成为ShuffleRDD,当调用mapPartitionsWithContext之后，shuffleRDD被转换成为MapPartitionsRDD
Log输出能证明我们的分析
res1: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[8] at reduceByKey at :13RDD转换小结
小结一下整个RDD转换过程
HadoopRDD->MappedRDD->FlatMappedRDD->MappedRDD->PairRDDFunctions->ShuffleRDD->MapPartitionsRDD
整个转换过程好长啊，这一切的转换都发生在任务提交之前。
所有的RDD操作都是在做逻辑上的计算，只有在通过下面讲的DAGScheduler的处理生成对应的stage并划分出具体的taskset的时候才会触发真正的的数据操作
运行过程分析
数据集操作分类
在对任务运行过程中的函数调用关系进行分析之前，我们也来探讨一个偏理论的东西，作用于RDD之上的Transformantion为什么会是这个样子？
对这个问题的解答和数学搭上关系了，从理论抽象的角度来说，任务处理都可归结为“input->processing->output"。input和output对应于数据集dataset.
在此基础上作一下简单的分类
1. one-one一个dataset在转换之后还是一个dataset，而且dataset的size不变，如map
2. one-one一个dataset在转换之后还是一个dataset，但size发生更改，这种更改有两种可能：扩大或缩小，如flatMap是size增大的操作，而subtract是size变小的操作
3. many-one多个dataset合并为一个dataset，如combine,
join4. one-many一个dataset分裂为多个dataset,如groupBy
Task运行期的函数调用
task的提交过程参考本系列中的第二篇文章。本节主要讲解当task在运行期间是如何一步步调用到作用于RDD上的各个operation
· TaskRunner.run
· Task.run
· Task.runTask (Task是一个基类，有两个子类，分别为ShuffleMapTask和ResultTask)
· RDD.iterator
· RDD.computeOrReadCheckpoint
· RDD.compute
或许当看到RDD.compute函数定义时，还是觉着f没有被调用,以MappedRDD的compute定义为例
overridedef compute(split: Partition, context: TaskContext) = firstParent[T].iterator(split, context).map(f)注意，这里最容易产生错觉的地方就是map函数，这里的map不是RDD中的map，而是scala中定义的iterator的成员函数map,请自行参考http://www.scala-lang.org/api/2.10.4/index.html#scala.collection.Iterator
堆栈输出
80 at org.apache.spark.rdd.HadoopRDD.getJobConf(HadoopRDD.scala:111) 81 at org.apache.spark.rdd.HadoopRDD$$anon$1.(HadoopRDD.scala:154) 82 at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:149) 83 at org.apache.spark.rdd.HadoopRDD.compute(HadoopRDD.scala:64) 84 at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241) 85 at org.apache.spark.rdd.RDD.iterator(RDD.scala:232) 86 at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31) 87 at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241) 88 at org.apache.spark.rdd.RDD.iterator(RDD.scala:232) 89 at org.apache.spark.rdd.FlatMappedRDD.compute(FlatMappedRDD.scala:33) 90 at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241) 91 at org.apache.spark.rdd.RDD.iterator(RDD.scala:232) 92 at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31) 93 at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241) 94 at org.apache.spark.rdd.RDD.iterator(RDD.scala:232) 95 at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:34) 96 at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241) 97 at org.apache.spark.rdd.RDD.iterator(RDD.scala:232) 98 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:161) 99 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:102) 100 at org.apache.spark.scheduler.Task.run(Task.scala:53) 101 at org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:211)ResultTask
compute的计算过程对于ShuffleMapTask比较复杂，绕的圈圈比较多，对于ResultTask就直接许多。
overridedef runTask(context: TaskContext): U = { metrics = Some(context.taskMetrics) try { func(context, rdd.iterator(split, context)) } finally { context.executeOnCompleteCallbacks() } }计算结果的传递
上面的分析知道，wordcount这个jo交之后，被DAGScheduler分为两个stage，第一个Stage是shuffleMapTask，第二个Stage是ResultTask.
那么ShuffleMapTask的计算结果是如何被ResultTask取得的呢？这个过程简述如下
1. ShffuleMapTask将计算的状态(注意不是具体的数据)包装为MapStatus返回给DAGScheduler（不可能包装真正的数据的）
2. DAGScheduler将MapStatus保存到MapOutputTrackerMaster中
3. ResultTask在执行到ShuffleRDD时会调用BlockStoreShuffleFetcher的fetch方法去获取数据（具体操作）：
1. 第一件事就是咨询MapOutputTrackerMaster所要取的数据的location
验证上述22. 根据返回的结果调用BlockManager.getMultiple获取真正的数据
BlockStoreShuffleFetcher的fetch函数伪码
val blockManager = SparkEnv.get.blockManager val startTime = System.currentTimeMillis val statuses = SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, reduceId) logDebug("Fetching map output location for shuffle %d, reduce %d took %d ms".format( shuffleId, reduceId, System.currentTimeMillis - startTime)) val blockFetcherItr = blockManager.getMultiple(blocksByAddress, serializer) val itr = blockFetcherItr.flatMap(unpackBlock)注意上述代码中的getServerStatuses及getMultiple,一个是询问数据的位置，一个是去获取真正的数据。