spark 调度模块详解及源码分析

@(SPARK)[spark]

spark 调度模块详解及源码分析
一概述
- 一三个主要的类
  - 1class DAGScheduler
  - 2trait TaskScheduler
  - 3trait SchedulerBackend
- 二基本流程
- 三TaskScheduler SchedulerBackend
二DAGScheduler
- 一用户代码中创建SparkContext对象SparkContext中创建DAGScheduler与TaskSchedulerTaskSchedulerBackend对象
  - 1用户代码中创建SparkContext对象
  - 2SparkContext源码简单分析
  - 3SparkContext创建DAGScheduler与TaskSchedulerTaskSchedulerBackend对象
  - 4createTaskScheduler创建TaskSchedulerTaskSchedulerBackend对象的具体过程
- 二用户代码创建各种transformation与至少一个action这个action会通过SparkContextrunJob调用DAGSchedulerrunJob
  - 1textFile
  - 2filter
  - 3count
- 三步骤三DAGScheduler提交作业划分stage并生成最终的TaskSet
  - 1创建DAGScheduler对象的详细实现
  - 2作业的提交
  - 3stage的划分
  - 4任务的生成
三TaskScheduler TaskSchedulerBackend

一、概述

通过spark-submit向集群提交应用后，spark就开始了调度的过程，其调度模块主要包括2部分：
* DAGScheduler：负责将用户提交的计算任务分割成不同的stage。
* TaskScheduler & SchedulerBackend：负责将stage中的task提交到集群中。
先看一下2个核心的图：

（一）三个主要的类

与调度模块相关的三个主要的类均位于org.apache.spark.scheduler，包括：

1、class DAGScheduler

负责分析用户提交的应用，并根据计算任务的依赖关系建立DAG，然后将DAG划分成不同的Stage，其中每个stage由可以并发执行的一组task构成，这些task的逻辑完全相同，只是作用于不同的数据。
DAGScheduler在不同的资源管理框架下的实现是完全相同的。

2、trait TaskScheduler

TaskScheduler从DAGScheduler中接收不同的stage的任务，并且向集群调度这些任务。
yarn-cluster的具体实现为：YarnClusterScheduler
yarn-client的具体实现为：YarnScheduler

3、trait SchedulerBackend

SchedulerBackend向当前等待分配资源的task分配计算资源，并且在分配的executor中启动task。
yarn-cluster的具体实现为：YarnClusterSchedulerBackend。
yarn-client的具体实现为：YarnClientSchedulerBackend

（二）基本流程

图片来源于spark内幕技术P45.

（三）TaskScheduler & SchedulerBackend

图片来源于spark内幕技术P44.
TaskScheduler侧重于调度，而SchedulerBackend是实际运行。

每个SchedulerBackend都会对应一个唯一的TaskScheduler。注意图中的TaskScheduler & SchedulerBackend都会有针对于yarn的特定实现。

二、DAGScheduler

DAGScheduler在不同的资源管理框架下的实现是完全相同的。因为DAGScheduler实现的功能是将DAG划分为不同的stage，这是根据宽依赖进行划分的，每个宽依赖均会调用shuffle，以此作为一个新的stage。这与具体的资源管理框架无关。

每个stage由可以并发执行的一组task构成，这些task的执行逻辑完全相同，只是作用于不同的数据。

DAGScheduler与TaskScheduler都是在SparkContext创建的时候创建的。其中TaskScheduler是通过SparkContext#createTaskScheduler创建的，而DAGScheduler是直接调用它的构造函数创建的。只不过，DAGScheduler保存了TaskScheduler的引用，因此需要先创建TaskScheduler。

＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝
步骤一：用户代码中创建SparkContext对象，SparkContext中创建DAGScheduler与TaskScheduler/TaskSchedulerBackend对象
步骤二：用户代码构建各种tranformation及至少一个action，这个action会通过SparkContext#runJob调用DAGScheduler#runJob
步骤三：DAGScheduler提交作业到一队列，handleJobSubmitted从这个队列取出作业，划分stage，并开始生成最终的TaskSet，调用submitTasks()向TaskScheduler提交任务

（一）用户代码中创建SparkContext对象，SparkContext中创建DAGScheduler与TaskScheduler/TaskSchedulerBackend对象

1、用户代码中创建SparkContext对象

下面我们从一个简单的程进行出发，分析它的进行过程。程序如下：

package com.lujinhong.sparkdemoimport org.apache.spark.SparkContext
object GrepWord {def grepCountLog(sc:SparkContext,path: String, keyWord: String) {println("grep " + keyWord + " in " + path + ", the lineCount is: ")val all = sc.textFile(path)val ret = all.filter(line => line.contains(keyWord))println(ret.count)}def main(args: Array[String]) {val sc = new SparkContext();grepCountLog(sc,"/src/20151201", "\"keyword\": \"20302\"");}
}

上面的代码很简单，指定一个目录，搜索这个目录中的文件有多少个keyword。分为三步：读入文件，过滤关键字，count。
我们这里主要分析yarn-cluster/client的模式，根据前面的分析，我们向YARN提交应用后，YARN会返回分配资源，然后启动AM。在AM中的driver会开始执行用户的代码，开始进行调度。详细分析请见：
http://blog.csdn.net/lujinhong2/article/details/50344095

那我们这里就从用户的代码开始继续往下分析。

用户代码中开始的时候必须首先创建一个SparkContext，我们看一下SparkContext的代码，以及它被创建时执行了哪些操作。

先看一下官方的一个图。DriverProgram就是用户提交的程序，在用户代码中创建一个SparkContext的对象。SparkContext是所有Spark应用的入口，它负责和整个集群的交互，包括创建RDD，累积器、广播变量等。

每个JVM中只能有一个SparkContext，在创建一个新的Context前，你必须先stop()旧的。这个限制可能会在以后去掉，见SPARK-2243。

2、SparkContext源码简单分析

SparkContext完成了以下几个主要的功能：
（1）创建RDD，通过类似textFile等的方法。
（2）与资源管理器交互，通过runJob等方法启动应用。
（3）创建DAGScheduler、TaskScheduler等。

3、SparkContext创建DAGScheduler与TaskScheduler/TaskSchedulerBackend对象

创建一个SparkContext，只需要一个SparkConf参数，表示一些配置项。如果未指定参数，则会创建一个默认的SparkConf，如我们代码中的：

val sc = new SparkContext();

创建的代码为：

def this() = this(new SparkConf())

在SparkContext的一个try模块中，会进行一些初始化的工作，其中一部分是创建了DAGScheduler与TaskScheduler/TaskSchedulerBackend对象。

// Create and start the scheduler
val (sched, ts) = SparkContext.createTaskScheduler(this, master)
_schedulerBackend = sched
_taskScheduler = ts
_dagScheduler = new DAGScheduler(this)// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's constructor
_taskScheduler.start()

其中TaskScheduler/TaskSchedulerBackend对象通过createTaskScheduler()方法进行创建，而DAGScheduler对象直接使用构建函数创建。

4、createTaskScheduler：创建TaskScheduler/TaskSchedulerBackend对象的具体过程

SparkContext通过createTaskScheduler来同时创建TaskScheduler/TaskSchedulerBackend对象。它接收的参数mater是一个url，或者一个yarn-cluster等的字符串，指明了使用哪种运行模式。

createTaskScheduler函数中主要就是match各种master，然后创建相应的TaskScheduler/TaskSchedulerBackend对象。

我们先看一下yarn-cluster的：

case "yarn-standalone" | "yarn-cluster" =>if (master == "yarn-standalone") {logWarning("\"yarn-standalone\" is deprecated as of Spark 1.0. Use \"yarn-cluster\" instead.")}val scheduler = try {val clazz = Utils.classForName("org.apache.spark.scheduler.cluster.YarnClusterScheduler")val cons = clazz.getConstructor(classOf[SparkContext])cons.newInstance(sc).asInstanceOf[TaskSchedulerImpl]} catch {// TODO: Enumerate the exact reasons why it can fail// But irrespective of it, it means we cannot proceed !case e: Exception => {throw new SparkException("YARN mode not available ?", e)}}val backend = try {val clazz =Utils.classForName("org.apache.spark.scheduler.cluster.YarnClusterSchedulerBackend")val cons = clazz.getConstructor(classOf[TaskSchedulerImpl], classOf[SparkContext])cons.newInstance(scheduler, sc).asInstanceOf[CoarseGrainedSchedulerBackend]} catch {case e: Exception => {throw new SparkException("YARN mode not available ?", e)}}scheduler.initialize(backend)(backend, scheduler)

TaskScheduler的实现类为：

org.apache.spark.scheduler.cluster.YarnClusterScheduler

TaskSchedulerBacked的实现类为：

org.apache.spark.scheduler.cluster.YarnClusterSchedulerBackend

其中后者需要前者来创建：

scheduler.initialize(backend)

最后返回一个元组：

(backend, scheduler)

再看看yarn-client的：

  case "yarn-client" =>val scheduler = try {val clazz = Utils.classForName("org.apache.spark.scheduler.cluster.YarnScheduler")val cons = clazz.getConstructor(classOf[SparkContext])cons.newInstance(sc).asInstanceOf[TaskSchedulerImpl]} catch {case e: Exception => {throw new SparkException("YARN mode not available ?", e)}}val backend = try {val clazz =Utils.classForName("org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend")val cons = clazz.getConstructor(classOf[TaskSchedulerImpl], classOf[SparkContext])cons.newInstance(scheduler, sc).asInstanceOf[CoarseGrainedSchedulerBackend]} catch {case e: Exception => {throw new SparkException("YARN mode not available ?", e)}}scheduler.initialize(backend)(backend, scheduler)

TaskScheduler的实现类为：

org.apache.spark.scheduler.cluster.YarnScheduler

TaskSchedulerBacked的实现类为：

org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend

（二）用户代码创建各种transformation与至少一个action，这个action会通过SparkContext#runJob调用DAGScheduler#runJob

在用户代码中创建了一个SparkContext对象后，就可以开始创建RDD，转换RDD等了。我们的代码只有3行：

val all = sc.textFile(path)
val ret = all.filter(line => line.contains(keyWord))
println(ret.count)

1、textFile

我们先看一下如何创建一个RDD:

  def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {assertNotStopped()hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],minPartitions).map(pair => pair._2.toString)}

读取一个hadoop支持的类型的文件，返回一个String类型的RDD。
它接收2个参数，一个是文件路径，一个是最小分区数，默认为：

def defaultMinPartitions: Int = math.min(defaultParallelism, 2)。

注意，这里使用的是min，因此如果没指定分区数量，最大的情况下就是2个分区了，详细分析请见： https://github.com/mesos/spark/pull/718

最后是调用hadoopFile来创建RDD的：

  def hadoopFile[K, V](path: String,inputFormatClass: Class[_ <: InputFormat[K, V]],keyClass: Class[K],valueClass: Class[V],minPartitions: Int = defaultMinPartitions): RDD[(K, V)] = withScope {assertNotStopped()// A Hadoop configuration can be about 10 KB, which is pretty big, so broadcast it.val confBroadcast = broadcast(new SerializableConfiguration(hadoopConfiguration))val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path)new HadoopRDD(this,confBroadcast,Some(setInputPathsFunc),inputFormatClass,keyClass,valueClass,minPartitions).setName(path)}

2、filter

仅返回符合条件的元素组成的RDD。

3、count

count的逻辑很简单：

  def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

sum用于统计一个集合中的元素数量。
但它调用了SparkContext的runJob开始执行任务了，我们分析一下这个过程。
SparkContext定义了多个runJob的形式，但它最后的调用为：

  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 = getCallSiteval 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()}

关键代码就是一行：

dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)

开始调用DAGScheduler的runJob了。

（三）步骤三：DAGScheduler提交作业，划分stage，并生成最终的TaskSet

经过上面的分析，我们知道了DAGScheduler与TaskScheduler/TaskSchedulerBackend对象的对象是如何创建的，并分析到了由用户代码出发，如果至调用dagScheduler.runJob。下面我们分析一下dagScheduler.runJob完成了什么功能。

1、创建DAGScheduler对象的详细实现

上面介绍过在SparkContext中会通过DAGScheduler的构建函数创建一个DAGScheduler对象，具体是如何实现的呢？

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 {......}

看一下DAGScheduler的主构造函数。
* SparkContext：就是前面创建的对象。
* taskScheduler：DAGScheduler保存一个taskScheduler，当最后处理完生成TaskSet时，需要调用submitMissingTasks，而在这个方法中会调用taskScheduler.submitTasks()，就是将TaskSet交由taskScheduler进行下一步的处理。
* mapOutputTracker：是运行在Driver端管理shuffle的中间输出位置信息的。
* blockManagerMaster：也是运行在Driver端的，它是管理整个Job的Bolck信息。

2、作业的提交

首先注意区分2个概述：
job: 每个action都是执行runJob方法，可以将之视为一个job。
stage：在这个job内部，会根据宽依赖，划分成多个stage。

前面说过，用户代码中存在一个action时，它最终会调用SparkContext#runJob()，而SparkContext#runJob()的最后一步都是调用DAGScheduler#runJob()

dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)

而DAGScheduler#runJob()的核心代码为：

val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)

即调用submitJob方法，我们进一步看看submitJob()

  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] = {
....    val jobId = nextJobId.getAndIncrement()
.....val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)eventProcessLoop.post(JobSubmitted(jobId, rdd, func2, partitions.toArray, callSite, waiter,SerializationUtils.clone(properties)))waiter}

submitJob()方法主要完成了以下3个工作：
* 获取一个新的jobId
* 生成一个JobWaiter，它会监听Job的执行状态，而Job是由多个Task组成的，因此只有当Job的所有Task均已完成，Job才会标记成功
* 最后调用eventProcessLoop.post()将Job提交到一个队列中，等待处理。这是一个典型的生产者消费者模式。这些消息都是通过handleJobSubmitted来处理。

简单看一下handleJobSubmitted是如何被调用的。
首先是DAGSchedulerEventProcessLoop#onReceive调用doOnReceive：

  /*** The main event loop of the DAG scheduler.*/override def onReceive(event: DAGSchedulerEvent): Unit = {val timerContext = timer.time()try {doOnReceive(event)} finally {timerContext.stop()}}

DAGSchedulerEventProcessLoop是EventLoop的子类，它重写了EventLoop的onReceive方法。以后再分析这个EventLoop。

然后，doOnReceive会调用handleJobSubmitted。

3、stage的划分

刚才说到handleJobSubmitted会从eventProcessLoop中取出Job来进行处理，处理的第一步就是将Job划分成不同的stage。handleJobSubmitted主要2个工作，一是进行stage的划分，这是这部分要介绍的内容；二是创建一个activeJob，并生成一个任务，这在下一小节介绍。

  private[scheduler] def handleJobSubmitted(jobId: Int,finalRDD: RDD[_],func: (TaskContext, Iterator[_]) => _,partitions: Array[Int],callSite: CallSite,listener: JobListener,properties: Properties) {...finalStage = newResultStage(finalRDD, partitions.length, jobId, callSite).....activeJobs += job......submitStage(finalStage)}submitWaitingStages()}

newResultStage()经过多层调用后，最终会调用getParentStages()。
因为是从最终的stage往回推算的，这需要计算最终stage所依赖的各个stage。

4、任务的生成

回到handleJobSubmitted中的代码：

 submitStage(finalStage)

submitStage会提交finalStage，如果这个stage的某些parentStage未提交，则递归调用submitStage()，直至所有的stage均已计算完成。

submitStage()会调用submitMissingTasks():

submitMissingTasks(stage, jobId.get)

而submitMissingTasks()会完成DAGScheduler最后的工作：它判断出哪些Partition需要计算，为每个Partition生成Task，然后这些Task就会封闭到TaskSet，最后提交给TaskScheduler进行处理。

taskScheduler.submitTasks(new TaskSet(tasks.toArray, stage.id, stage.latestInfo.attemptId, stage.firstJobId, properties))stage.latestInfo.submissionTime = Some(clock.getTimeMillis())

三、TaskScheduler && TaskSchedulerBackend

上文分析到在DAGScheduler中最终会执行taskScheduler.submitTasks()方法，我们先简单看一下从这里开始往下的执行逻辑：

（1）taskScheduler#submitTasks()
（2） schedulableBuilder#addTaskSetManager()
（3）CoarseGrainedSchedulerBackend#reviveOffers()
（4）CoarseGrainedSchedulerBackend#makeOffers()
（5）TaskSchedulerImpl#resourceOffers
（6）CoarseGrainedSchedulerBackend#launchTasks
（7）executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))

步骤一、二中主要将这组任务的TaskSet加入到一个TaskSetManager中。TaskSetManager会根据数据就近原则为task分配计算资源，监控task的执行状态等，比如失败重试，推测执行等。
步骤三、四逻辑较为简单。
步骤五为每个task具体分配资源，它的输入是一个Executor的列表，输出是TaskDescription的二维数组。TaskDescription包含了TaskID, Executor ID和task执行的依赖信息等。
步骤六、七就是将任务真正的发送到executor中执行了，并等待executor的状态返回。