内容：
1、再次思考pipeline；
2、窄依赖物理执行内幕；
3、宽依赖物理执行内幕；
4、Job提交流程；
物理执行是更深层次的角度。
==========再次思考pipeline ============
即使采用pipeline的方式，函数f对依赖的RDD中的数据集合的操作也会有两种方式：
1、f(record)，f作用于集合的每一条记录，每次只作用于一条记录；
2、f(records)， f一次性作用于集合的全部数据；
Spark运行的时候用的是第一种方式，为什么呢？
1、无需等待，可以最大化的使用集群的计算资源；
2、减少OOM的发生；
3、最大化的有利于并发；
4、可以精准的控制每一个partition本身(Dependency)极其内部的计算(compute)；
5、基于lineage的算子流动式函数式编程，节省了中间结果的产生，并且可以最快的恢复；
疑问：会不会增加网络通信？
当然不会！因为在pipeline！把很多算子合并成一个算子，在一个stage。
==========思考Spark Job的物理执行============
Spark Application里面可以产生一个或者多个Job，例如Spark Shell默认启动的时候内部就没有Job，只是作为资源的分配程序，可以在Spark Shell里面写代码产生若干个Job。普通程序中一般而言可以有不同的Action，每一个Action一般也会触发一个Job。
Spark是MapReduce思想的一种更加精致和高效的实现。MapReduce有很多具体不同的实现。
例如：Hadoop的MapReduce基本的计算流程如下：首先是以JVM为对象的并发执行的Mapper，Mapper中的map的执行会产生输出数据，输出的数据会经由partitioner指定的规则放到LocalFile System中，然后再经由Shuffle、Sort、Aggregate变成reducer中的reduce的输入，执行reduce产生最终的执行结果。Hadoop MapReduce执行的流程虽然简单，但是过于死板，尤其是在构造复杂（迭代）算法的时候，非常不利于算法的实现，且执行效率极为低下。
Spark算法构造和物理执行时最最基本的核心：最大化pipeline。pipeline 越多，复用越好。基于pipeline的思想是数据被使用的时候才开始计算。从数据流动流动角度来说，是数据流动到计算的位置！！！实质上从逻辑的角度来看，是算子在数据上流动。
从算法构建角度而言，肯定是算子作用于数据，所以是算子在数据上流动，有利于构建算法。从物理执行的角度而言，是数据流动到计算的位置，有利于系统最高效的计算。
数据要流动到计算的位置，那这个位置在哪个地方？
对于pipeline而言，数据计算的位置就是每个stage位置中最后的RDD。就算一个stage有5000个步骤，真正的计算，也还是在第5000步计算(比如第5000个函数！！！)，发给executor之前，所有的算子已经合并成1个了。如果第5步需要cache一下，是系统自己的设计，和前面一段话无关。
一个震撼人心的内幕真相就是：每个stage，除了最后一个RDD算子是真实的之外，前面的算子都是假的。
由于计算的lazy特性，导致计算从后往前回溯，形成Computing Chain，导致的结果就是：需要首先计算出具体一个stage内部最左侧的RDD中本次计算依赖的partition
如果stage没有parent stage的话，则stage从最左边的RDD立即执行，每次计算出的record都流入下一个函数。
计算步骤从后往前回溯，计算执行是从前往后。
==========窄依赖物理执行内幕 ============
一个Stage内部的RDD都是窄依赖，窄依赖从逻辑上看是从内幕最左侧的RDD开始立即计算的。
根据计算链条，数据从一个计算步骤流动到下一个计算步骤，以此类推，直到计算到Stage内部的最后一个RDD来产生计算结果。
Computing Chain构建是从后往前回溯而成的，而实际的物理计算则是让数据从前往后在算子上流动，直到流动到不能再流动为止才开始计算下一个record。
所有RDD里面的compute都是iterator来一步步执行。
后面的RDD对前面的RDD的依赖虽然是partition级别的依赖，但是并不需要父RDD把partition中所有的records计算完毕才整体往后流动数据进行计算，这就极大的提高了计算速率。
/**
* An RDD that applies the provided function to every partition of the parent RDD.
*/
private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag](
prev: RDD[T],
f: (TaskContext, Int, Iterator[T]) => Iterator[U], // (TaskContext, partition index, iterator)
preservesPartitioning: Boolean = false)
extends RDD[U](prev) {
override val partitioner = if (preservesPartitioning) firstParent[T].partitioner else None
override def getPartitions: Array[Partition] = firstParent[T].partitions
override def compute(split: Partition, context: TaskContext): Iterator[U] =
f(context, split.index, firstParent[T].iterator(split, context))
}
==========宽依赖物理执行内幕 ============
必须等到依赖的父Stage中的最后一个RDD把全部数据彻底计算完毕才能够经过shuffle来计算当前的Stage。
这样写代码的时候尽量避免宽依赖！！！
/**
* Implemented by subclasses to return how this RDD depends on parent RDDs. This method will only
* be called once, so it is safe to implement a time-consuming computation in it.
*/
protected def getDependencies: Seq[Dependency[_]] = deps
compute负责接受父Stage的数据流，计算出record
==========Job提交流程 ============
作业提交，触发Action
/**
* Run a function on a given set of partitions in an RDD and pass the results to the given
* handler function. This is the main entry point for all actions in Spark.
*/
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
resultHandler: (Int, U) => Unit): Unit = {
if (stopped.get()) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
if (conf.getBoolean("spark.logLineage", false)) {
logInfo("RDD‘s recursive dependencies:\n" + rdd.toDebugString)
}
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
rdd.doCheckpoint()
}
/**
* Run an action job on the given RDD and pass all the results to the resultHandler function as
* they arrive.
*
* @param rdd target RDD to run tasks on
* @param func a function to run on each partition of the RDD
* @param partitions set of partitions to run on; some jobs may not want to compute on all
* partitions of the target RDD, e.g. for operations like first()
* @param callSite where in the user program this job was called
* @param resultHandler callback to pass each result to
* @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
*
* @throws Exception when the job fails
*/
def runJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): Unit = {
val start = System.nanoTime
val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
waiter.awaitResult() match {
case JobSucceeded =>
logInfo("Job %d finished: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
case JobFailed(exception: Exception) =>
logInfo("Job %d failed: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
// SPARK-8644: Include user stack trace in exceptions coming from DAGScheduler.
val callerStackTrace = Thread.currentThread().getStackTrace.tail
exception.setStackTrace(exception.getStackTrace ++ callerStackTrace)
throw exception
}
}
/**
* Submit an action job to the scheduler.
*
* @param rdd target RDD to run tasks on
* @param func a function to run on each partition of the RDD
* @param partitions set of partitions to run on; some jobs may not want to compute on all
* partitions of the target RDD, e.g. for operations like first()
* @param callSite where in the user program this job was called
* @param resultHandler callback to pass each result to
* @param properties scheduler properties to attach to this job, e.g. fair scheduler pool name
*
* @return a JobWaiter object that can be used to block until the job finishes executing
* or can be used to cancel the job.
*
* @throws IllegalArgumentException when partitions ids are illegal
*/
def submitJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
// Check to make sure we are not launching a task on a partition that does not exist.
val maxPartitions = rdd.partitions.length
partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
throw new IllegalArgumentException(
"Attempting to access a non-existent partition: " + p + ". " +
"Total number of partitions: " + maxPartitions)
}
val jobId = nextJobId.getAndIncrement()
if (partitions.size == 0) {
// Return immediately if the job is running 0 tasks
return new JobWaiter[U](this, jobId, 0, resultHandler)
}
assert(partitions.size > 0)
val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))
waiter
}
作业：
写一下我理解中的spark job物理执行。