Spark Runtime里的主要层次分析，梳理Runtime组件和执行流程，
DAGScheduler
Job=多个stage，Stage=多个同种task, Task分为ShuffleMapTask和ResultTask，Dependency分为ShuffleDependency和NarrowDependency
面向stage的切分，切分依据为宽依赖
维护waiting jobs和active jobs，维护waiting stages、active stages和failed stages，以及与jobs的映射关系
主要职能
接收提交Job的主入口，submitJob(rdd, …)或runJob(rdd, …)。在SparkContext里会调用这两个方法。 处理TaskCompletionEvent 如果task执行成功，对应的stage里减去这个task，做一些计数工作： 其他与job相关的操作还包括：cancel job， cancel stage, resubmit failed stage等
其他职能 1. cacheLocations 和 preferLocation
private val cacheLocs = new HashMap[Int, Array[Seq[TaskLocation]]]TaskScheduler
维护task和executor对应关系，executor和物理资源对应关系，在排队的task和正在跑的task。
内部维护一个任务队列，根据FIFO或Fair策略，调度任务。
TaskScheduler本身是个接口，spark里只实现了一个TaskSchedulerImpl，理论上任务调度可以定制。下面是TaskScheduler的主要接口：
def start(): Unitdef postStartHook() { }def stop(): Unitdef submitTasks(taskSet: TaskSet): Unitdef cancelTasks(stageId: Int, interruptThread: Boolean)def setDAGScheduler(dagScheduler: DAGScheduler): Unitdef executorHeartbeatReceived(execId: String, taskMetrics: Array[(Long, TaskMetrics)],blockManagerId: BlockManagerId): Boolean
主要职能
submitTasks(taskSet)，接收DAGScheduler提交来的tasks 为tasks创建一个TaskSetManager，添加到任务队列里。TaskSetManager跟踪每个task的执行状况，维护了task的许多具体信息。触发一次资源的索要。 首先，TaskScheduler对照手头的可用资源和Task队列，进行executor分配(考虑优先级、本地化等策略)，符合条件的executor会被分配给TaskSetManager。然后，得到的Task描述交给SchedulerBackend，调用launchTask(tasks)，触发executor上task的执行。task描述被序列化后发给executor，executor提取task信息，调用task的run()方法执行计算。cancelTasks(stageId)，取消一个stage的tasks 调用SchedulerBackend的killTask(taskId, executorId, …)方法。taskId和executorId在TaskScheduler里一直维护着。resourceOffer(offers: Seq[Workers])，这是非常重要的一个方法，调用者是SchedulerBacnend，用途是底层资源SchedulerBackend把空余的workers资源交给TaskScheduler，让其根据调度策略为排队的任务分配合理的cpu和内存资源，然后把任务描述列表传回给SchedulerBackend 从worker offers里，搜集executor和host的对应关系、active executors、机架信息等等worker offers资源列表进行随机洗牌，任务队列里的任务列表依据调度策略进行一次排序遍历每个taskSet，按照进程本地化、worker本地化、机器本地化、机架本地化的优先级顺序，为每个taskSet提供可用的cpu核数，看是否满足 statusUpdate(taskId, taskState, data),另一个非常重要的方法，调用者是SchedulerBacnend，用途是SchedulerBacnend会将task执行的状态汇报给TaskScheduler做一些决定
TaskResultGetter取task result的逻辑
对于success task，如果taskResult里的数据是直接结果数据，直接把data反序列出来得到结果；如果不是，会调用blockManager.getRemoteBytes(blockId)从远程获取。如果远程取回的数据是空的，那么会调用TaskScheduler.handleFailedTask，告诉它这个任务是完成了的但是数据是丢失的。否则，取到数据之后会通知BlockManagerMaster移除这个block信息，调用TaskScheduler.handleSuccessfulTask，告诉它这个任务是执行成功的，并且把result data传回去。对于failed task，从data里解析出fail的理由，调用TaskScheduler.handleFailedTask，告诉它这个任务失败了，理由是什么。SchedulerBackend
在TaskScheduler下层，用于对接不同的资源管理系统，SchedulerBackend是个接口，需要实现的主要方法如下：
def start(): Unitdef stop(): Unitdef reviveOffers(): Unit // 重要方法：SchedulerBackend把自己手头上的可用资源交给TaskScheduler，TaskScheduler根据调度策略分配给排队的任务吗，返回一批可执行的任务描述，SchedulerBackend负责launchTask，即最终把task塞到了executor模型上，executor里的线程池会执行task的run()def killTask(taskId: Long, executorId: String, interruptThread: Boolean): Unit =throw new UnsupportedOperationException
粗粒度：进程常驻的模式，典型代表是standalone模式，mesos粗粒度模式，yarn
细粒度：mesos细粒度模式
这里讨论粗粒度模式，更好理解：CoarseGrainedSchedulerBackend。
维护executor相关信息(包括executor的地址、通信端口、host、总核数，剩余核数)，手头上executor有多少被注册使用了，有多少剩余，，总共还有多少核是空的等等。
主要职能
Driver端主要通过actor监听和处理下面这些事件： reviveOffers()方法的实现。直接调用了makeOffers()方法，得到一批可执行的任务描述，调用launchTasks。launchTasks(tasks: Seq[Seq[TaskDescription]])方法。 遍历每个task描述，序列化成二进制，然后发送给每个对应的executor这个任务信息 如果这个二进制信息太大，超过了9.2M(默认的akkaFrameSize 10M 减去 默认 为akka留空的200K)，会出错，abort整个taskSet，并打印提醒增大akka frame size如果二进制数据大小可接受，发送给executor的actor，处理LaunchTask(serializedTask)事件。Executor
Executor是spark里的进程模型，可以套用到不同的资源管理系统上，与SchedulerBackend配合使用。
内部有个线程池，有个running tasks map，有个actor，接收上面提到的由SchedulerBackend发来的事件。
事件处理
launchTask。根据task描述，生成一个TaskRunner线程，丢尽running tasks map里，用线程池执行这个TaskRunnerkillTask。从running tasks map里拿出线程对象，调它的kill方法。
全文完 🙂
不论你在什么时候开始，重要的是开始之后就不要停止
