Spark源码分析心得

要想学习理解一款流行分布式系统的源码不是一件容易的事情，必定要屡次迭代，看无数遍而且领悟其设计思想。第一次看不要纠结于细节，每次迭代过程当中增长一点点细节的理解，最终达到豁然开朗的地步。

学习优秀的源代码是提升自身技能的最好途径，比作无数个低水平的项目效果要显著的多，好了，闲话少说，让咱们试图来理解Spark的世界吧。

一、大框架

首先要掌握几个基本概念，Spark是分布式计算框架，核心思想是经过将计算任务尽可能分配到源数据一致的机器上执行，下降网络延时；同时引入Dag图依赖关系生成一系列计算任务，固然缓存等机制是不可避免要用到的，为了提升性能嘛。

大部分分布式计算的核心思想是相似的，经过成熟的分布式框架在集群间通讯和同步消息，提供横向扩展能力，知足大数据计算的需求。在Spark中具体的分布式消息传递是经过Akka模块来支持的。

核心类：

SparkContext，DagSchedule，TaskScheduleImp（TaskSchedule的实现），Stage，Task，TaskDescription，TaskInfo，RDD，BlockManager等

（1）SparkContext

建立spark任务主要是经过它来完成的，Spark程序上下文，里面包含了DagSchedule，TaskSchedule，BlockManager等等。

（2）DagSchedule

矢量图计算，解析整个Spark任务生成Stage调用树，每一个Stage的划分主要是看该Stage是否包含Shuffle过程来决定，Stage在调用的时候生成TaskSet，并经过TaskSchedule分配到具体的Executor上，每一个TaskSet包含1到多个Task，Task具体分红ResultTask和ShuffleResultTask两种，两种的区别从命名上就能够区分出来，前者直接计算获得结果，后者是和Shuffle过程相关的。

（3）TaskScheduleImp

经过具体的TaskScheduleEndpoint在集群间通讯，发布任务等。

与集群通讯是经过TaskScheduleEndpoint来执行的。最经常使用的是CoarseGrainedSchedulerBackend

（4）Executor

执行单元，通常每台集群机器会分配一个Executor，每一个Executor管理本地的多个TaskSet。

Executor经过ExecutorBackend来和Master的ScheduleEndpoint通讯，相应的最经常使用的Endpoint是CoarseGrainedExecutorBackend。

（5）RDD

数据集，Spark定义了不少的数据集（RDD），好比HadoopRDD，JdbcRDD等。

RDD中的具体的数据有时是经过BlockManager来管理的，RDD中能寻址到数据所在的机器。

（6）Task

具体的任务，Master定义好Task后发布到集群中对应机器的Exector去执行，执行结果经过DirectResult和IndirectResult返回，后者经过包含告终果数据所在的Shuffle地址或者块地址等寻址信息。

（7）BlockManager

管理整个集群的Block，默认Block大小是128M，内存和磁盘数据的对应关系等也是经过相关的类来管理的。

以上是初步的比较笼统的一个框架结构，主要用于增强理解，要想更好的理解Spark必需要经过不断的读源码，后续时间笔者会依次和你们分享更具体的源码心得。

（8）MapOutTracker

跟踪整个集群的MapStatus，不一样集群之间经过MapOutTrackerMaster等通讯来同步信息。

 

二、DagSchedule任务调度

class DAGScheduler(
    private[scheduler] val sc: SparkContext,
    private[scheduler] val taskScheduler: TaskScheduler,
    listenerBus: LiveListenerBus,
    mapOutputTracker: MapOutputTrackerMaster,
    blockManagerMaster: BlockManagerMaster,
    env: SparkEnv,
    clock: Clock = new SystemClock())
  extends Logging

（1）Dag接受到的各类命令都经过Dag内部事件的方式被执行（而不是直接执行），便于Dag内部作一些排序调度的判断准备工做。

命令的种类：心跳、Exector的注册注销、Task的执行结束失败

（2）两种主要的stage：ResultStage，ShuffleMapStage

getShuffleMapStage：生成或者得到Shuffle类型的stage，同时要生成Shuffle的locs位置信息在里面

getParentStage：根据Rdd各依赖Rdd的Shuffle属性，获取全部Stage列表，这些Stage执行完成以后才能执行本Stage。

getMissingParentStages：获取全部缺失的父Stage，这里只判断Shuffle Stage的状况，ResultStage不考虑，若是缺失则生成该ShuffleToMapStage。

其余一系列维护Stage、Job关系的属性和方法。

（3）主要的Submit方式

SubmitJob：提交做业

生成该做业的finalStage，而后再生成ActiveJob，组装各状态HashMap,提交该finalStage

SubmitStage：提交某个Stage，与SubmitJob的区别是不生成新的ActiveJob。

submitMissingTasks：真正的提交Stage，当全部父Stage都准备就绪时执行，要重点看。stage的全部partitions生成多个Task。最后将这些tasks合并成TaskSet并提交到TaskSchedule（taskScheduler.submitTasks( new TaskSet(...）)).

（4）TaskCompleted事件处理

广播事件到listenerBus；

找到Task所在的Stage：

a、若是是ResultTask，则更新该Stage属于的Job的状态，并判断Stage是否全部Task都执行完成，若是是则触发StageCompleted事件。好像也多是JobCompleted事件，还要再看一次。

b、ShuffleMapTask，则找到对应的ShuffleStage，更新对应的Task所在分区的MapStatus（或者位置信息等），更新mapOutputTracker状态属性，最后启动全部知足提交状态的等待Stage（waitingStages）。

（5）handleExecutorAdded

当有额外的Executor加进来的时候，只是执行SubmitWaitingStage命令，这时候可能会有等待Stage知足了执行条件。

 

三、Task

ResultTask：直接在RDD上执行func

ShuffleMapTask：执行ShuffleDependcy的shuffleHandler方法，主要是一些聚合函数的计算，而后将对应的RDD分区数据执行这些聚合计算以后的结果写入到shuffleManager管理的writer中去，多是写入到内存也可能写入到disk（猜想），shuffleManager通常会用到BlockManager来管理数据的存储。

Shuffle结果通常存储到集群的临时目录中，具体规则可参见DiskBlockManager和FileShuffleBlockResolver等类实现。

具体内容参见代码。

shuffleManager有这么几种：

（1）HashShuffleManager

（2）SortShuffleManager

固然也能够自定义