DAGScheduler
面向stage的调度层，为job生成以stage组成的DAG，提交TaskSet给TaskScheduler运行。
每个Stage内，都是独立的tasks，他们共同运行同一个compute function，享有同样的shuffledependencies。DAG在切分stage的时候是按照出现shuffle为界限的。
private[spark]class DAGScheduler( taskScheduler: TaskScheduler, listenerBus: LiveListenerBus, mapOutputTracker: MapOutputTrackerMaster, blockManagerMaster: BlockManagerMaster, env: SparkEnv) extends Logging { // Actor模式收取发来的DAGSchedulerEvent，并进行processEventprivate var eventProcessActor: ActorRef = _ private[scheduler] val nextJobId = new AtomicInteger(0) private[scheduler] def numTotalJobs: Int = nextJobId.get() private val nextStageId = new AtomicInteger(0) // 一系列信息维护，非常清晰 private[scheduler] val jobIdToStageIds = new HashMap[Int, HashSet[Int]] private[scheduler] val stageIdToJobIds = new HashMap[Int, HashSet[Int]] private[scheduler] val stageIdToStage = new HashMap[Int, Stage] private[scheduler] val shuffleToMapStage = new HashMap[Int, Stage] private[scheduler] val jobIdToActiveJob = new HashMap[Int, ActiveJob] private[scheduler] val resultStageToJob = new HashMap[Stage, ActiveJob] private[scheduler] val stageToInfos = new HashMap[Stage, StageInfo] // 不同状态stages的维护，非常清晰 // Stages we need to run whose parents aren't done private[scheduler] val waitingStages = new HashSet[Stage] // Stages we are running right now private[scheduler] val runningStages = new HashSet[Stage] // Stages that must be resubmitted due to fetch failures private[scheduler] val failedStages = new HashSet[Stage] // Missing tasks from each stage private[scheduler] val pendingTasks = new HashMap[Stage, HashSet[Task[_]]] private[scheduler] val activeJobs = new HashSet[ActiveJob] // Contains the locations that each RDD's partitions are cached on private val cacheLocs = new HashMap[Int, Array[Seq[TaskLocation]]]
在start()方法中会初始化Actor，然后接收DAGSchedulerEvent处理。Scheduler会在SparkContext里start起来。
Event处理
源代码的阅读入口：能够依据processEvent(event:DAGSchedulerEvent)方法展开。
处理的事件包含这么一些：
Submit Job
JobSubmitted事件：
提交任务的事件传入參数例如以下
case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties)
处理过程能够拆成三步看，每一步里面涉及的详细逻辑在以下补充展开
finalStage = newStage(rdd, partitions.size, None, jobId, Some(callSite))
本次newStage()操作能够相应新的result stage或者shuffle stage。返回Stage类(里面记录一些信息)。Stage类会传入Option[ShuffleDependency[_,_]]參数，内部有一个isShuffleMap变量，以标识该Stage是shuffle or result。
val job = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)
ActiveJob类也是记录一些信息的类，能够当作是一个VO类
if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {// Compute very short actions like first() or take() // with no parent stages locally.listenerBus.post(SparkListenerJobStart(job.jobId, Array[Int](), properties))runLocally(job)} else {jobIdToActiveJob(jobId) = jobactiveJobs += jobresultStageToJob(finalStage) = joblistenerBus.post(SparkListenerJobStart(job.jobId, jobIdToStageIds(jobId).toArray, properties))submitStage(finalStage)} 首先推断stage没有父亲依赖，且partition为1的话，就运行本地任务。否则，submitStage。
submitStage的逻辑为，首先寻找本次stage的parents。假设没有missing的parent stage，那么就submitMissingTask,即提交本次stage的tasks了。假设有，会对parent stage进行递归submitStage，并且getMissingParentStages得到的结果集是按id降序排的，也就是说递归submitStage的时候会按parent stage的id顺序进行。
submitMissingTask处理的是stage的parent已经available的stage。主要逻辑例如以下：
第一步：通过stage.isShuffleMap来决定生成ShuffleMapTask还是ResultTask，生成的ShuffleMapTask数目和partition数目相等。
第二步：把生成的tasks组建成一个TaskSet，提交给TaskScheduler的submitTasks方法。
TaskScheduler
DAGScheduler以stage为单位，提tasks给TaskScheduer，实现类为TaskSchedulerImpl。
TaskSchedulerImpl几个内部部件：
SchedulerBackend
SchedulableBuilder
DAGScheduler
TaskSetManager
TaskResultGetter
Tasks信息（taskIdToTaskSetId，taskIdToExecutorId，activeExecutorIds）
别的信息（SchedulerMode）
TaskScheduler做接收task、接收分到的资源和executor、维护信息、与backend打交道、把任务分配好等事情。
start()，stop()的时候，backend的start()，stop()
submitTasks(TaskSet)逻辑：
为这批Task生成新的TaskSetManager，把TaskSetManager加到SchedulerBuilder里，然后向backend进行一次reviveOffer()操作。
SchedulerBuilder
SchedulableBuilder有FIFO和Fair两种实现， addTaskSetManager会把TaskSetManager加到pool里。FIFO的话仅仅有一个pool。Fair有多个pool，Pool也分FIFO和Fair两种模式。
SchedulableBuilder的rootPool里面能够新增pool或者TaskSetManager，两者都是Scheduable的继承类，所以SchedulableBuilder用于维护rootPool这棵Scheduable 树结构。Pool是树上的非叶子节点，而TaskSetManager就是叶子节点。
在TaskScheduler初始化的时候会buildDafaultPool。
TaskSetManager
TaskSetManager负责这批Tasks的启动，失败重试，感知本地化等事情。每次reourseOffer方法会寻找合适(符合条件execId, host, locality)的Task并启动它。
reourseOffer方法，
def resourceOffer( execId: String, host: String, maxLocality: TaskLocality.TaskLocality)
寻找符合execId, host和locality的task，找到的话就启动这个Task。启动的时候，把task加到runningTask的HashSet里，然后调DAGScheduler的taskStarted方法，taskStarted方法向eventProcessorActor发出BeginEvent的DAGSchedulerEvent。
TaskResultGetter
维护一个线程池，用来反序列化和从远端获取task结果。
def enqueueSuccessfulTask(taskSetManager: TaskSetManager, tid: Long, serializedData: ByteBuffer)
把序列化的数据反序列化解析出来之后，有两种情况：直接可读的result和间接task result。
前者是DirectTaskResult[T]类：
class DirectTaskResult[T](var valueBytes: ByteBuffer, var accumUpdates: Map[Long, Any], var metrics: TaskMetrics)
后者是IndirectTaskResult[T]类：
case class IndirectTaskResult[T](blockId: BlockId) extends TaskResult[T] with Serializable
在反解析出IndirectTaskResult后，能够得到BlockId这个类，他的实现有这么些：
在TaskResultGetter里，会通过blockManager的getRemoteBytes(BlockId)方法来获得序列化的task result，对这个task result进行反解析后得到DirectTaskResult类，从而获得反序列化后的真正结果数据。
这是大致的一个过程，详细另一些细节在之中，比方会向scheduler发送不同的event、blockManager会调用BlockManagerMaster把该Block remove掉。
BlockId类有这么些关键变量：
private[spark] sealed abstract class BlockId { /** A globally unique identifier for this Block. Can be used for ser/de. */ def name: String // convenience methods def asRDDId = if (isRDD) Some(asInstanceOf[RDDBlockId]) else None def isRDD = isInstanceOf[RDDBlockId] def isShuffle = isInstanceOf[ShuffleBlockId] def isBroadcast = isInstanceOf[BroadcastBlockId] 以下看BlockManager怎样通过BlockId获得数据：
调用的是BlockManager的内部方法
private def doGetRemote(blockId: BlockId, asValues: Boolean): Option[Any] = {require(blockId != null, "BlockId is null")// 通过BlockManagerMaster获得这个blockId的locations val locations = Random.shuffle(master.getLocations(blockId)) for (loc <- locations) { logDebug("Getting remote block " + blockId + " from " + loc) // 使用BlockManagerWorker来获得block的数据 val data = BlockManagerWorker.syncGetBlock( GetBlock(blockId), ConnectionManagerId(loc.host, loc.port)) if (data != null) { if (asValues) { // 取到就返回 return Some(dataDeserialize(blockId, data)) } else { return Some(data) } } logDebug("The value of block " + blockId + " is null") } logDebug("Block " + blockId + " not found") None } 思路是通过BlockManagerMaster来获得block的位置信息，得到的集合打乱后，遍历位置信息，通过BlockManagerWorker去获得数据，仅仅要得到了，就反序列化之后返回。
在TaskResultGetter处理的时候，成功和失败分别向Scheduler调用handleSuccessfulTask和handleFailedTask方法。
handleSuccessfulTask在DAGScheduler里，会发出CompletionEvent事件，这一步结尾工作会有非常多细节处理，这里先不阅读了。
handleFailedTask的话，仅仅要TaskSetManager不是zombie，task没有被kill，那么会继续调用backend.reviveOffers()来re-run。
全文完 :)