Spark-Job启动、Stage划分

一、上下文

《Spark-driver和executor启动过程》详细分析了driver和executor的启动，此时资源已经给我们分配好了，且Application也已经注册完成。下面我们就来看看Spark是如何启动job并根据DAG来划分Stage的

二、Job启动

Spark RDD中的算子分为Transformations 算子和Actions 算子，Transformations 算子只是将RDD在逻辑上进行了转换，只有调用Actions 算子时才会真正执行以上对RDD的所有操作。为什么呢？点进去这些Actions 算子看下就发现它们都调用了SparkContext的runJob() ，因此程序调用几次Actions算子就会启动几个Job，因此一个Application是对应多个Job的。下面继续看下runJob()中做了什么事情。

1、SparkContext

class SparkContext(config: SparkConf) extends Logging {//在RDD中的给定分区集上运行一个函数，并将结果传递给给定的处理程序函数。//这是Spark中所有操作的主要入口点。def runJob[T, U: ClassTag](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],resultHandler: (Int, U) => Unit): Unit = {......//打印启动job的日志logInfo("Starting job: " + callSite.shortForm)//调用dagScheduler的runJob()dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)progressBar.foreach(_.finishAll())//在使用此RDD的作业完成后调用它rdd.doCheckpoint()}}

2、DAGScheduler

DAGScheduler 是面向 stage 调度的高级调度层。它为每个作业计算一个stage的DAG，跟踪这些RDD和stage输出被物化，并找到运行作业的最小时间表。然后，它将stage作为TaskSet（包含完全独立的任务）提交给在集群上运行它们的底层TaskScheduler实现。

Spark stages 是通过在Shuffle边界处打破RDD图来创建的。是一系列具有“窄”依赖关系的RDD操作，如map() 和filter()，这些窄依赖的RDD在每个 stage 都被 pipelined 到一组任务中，但具有shuffle依赖关系的操作需要多个stage（一个stage shuffle write 一组文件，另一个stage shuffle read 这些文件）。最后，每个stage都只会对其他stage进行shuffle依赖，并可能在其中计算多个操作。这些操作的实际 pipelining 发生在各种RDD的RDD.compute()函数中

除了生成阶段的DAG外，DAGScheduler还根据当前缓存状态确定运行每个 Task 的首选位置，并将其传递给低级TaskScheduler。此外，它还可以处理由于shuffle输出文件丢失而导致的故障，在这种情况下，可能需要重新提交之前跑过的stage。非随机文件丢失导致的阶段内故障由TaskScheduler处理，TaskScheduler将在取消整个stage之前重试每个任务几次。

在看源码时，有几个关键概念：

Jobs

是提交给调度器的顶级工作项，例如，当用户调用类似count() 的操作时，作业将通过submitJob提交。每个Job可能需要执行多个Stage来构建中间数据。

Stages

是计算Job中中间结果的一组任务，其中每个Task在同一RDD的分区上计算相同的函数。
Stage在shuffle边界处分开，也说明我们必须等待前一阶段完成才能获取输出。有两种类型的阶段：ResultStage，用于执行动作的最后阶段，以及ShuffleMapStage，用于为shuffle写入映射输出文件。如果这些作业重用相同的RDD，则Stage通常在多个Job之间共享。

Tasks

单独的工作单元，每个任务都发送到一台机器上。

Cache tracking

DAGScheduler计算出哪些RDD被缓存以避免重新计算它们，并且同样记住哪些shuffle write阶段已经生成了输出文件，以避免重做shuffle write

Preferred locations

DAGScheduler还根据其底层RDD的首选位置，或缓存或shuffle数据的位置，计算在stage中中的每个Task应该去哪台节点执行

Cleanup

当Job完成后，所有依赖的数据结构都会被清除，防止OOM

private[spark] 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 {def runJob[T, U](...) {//提交要给Job给调度器val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)ThreadUtils.awaitReady(waiter.completionFuture, Duration.Inf)}}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] = {// 获取rdd的所有分区数量val maxPartitions = rdd.partitions.length......//等待DAGScheduler作业完成的对象。当任务完成时，它将其结果传递给给定的处理函数。val waiter = new JobWaiter[U](this, jobId, partitions.size, resultHandler)//向线程组提交一个 JobSubmitted 事件//在目标RDD上提交了一个产生结果的作业eventProcessLoop.post(JobSubmitted(jobId, rdd, func2, partitions.toArray, callSite, waiter,Utils.cloneProperties(properties)))waiter}}private[scheduler] class DAGSchedulerEventProcessLoop(dagScheduler: DAGScheduler)extends EventLoop[DAGSchedulerEvent]("dag-scheduler-event-loop") with Logging {private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>//处理 JobSubmitted 事件 转交给 dagScheduler 处理dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)}}

三、Stage划分

Stage是一组并行Task，它们都计算需要作为Spark作业的一部分运行的相同函数，
其中所有Task都具有相同的shuffle依赖关系。调度器运行的每个Task
DAG在Shuffle发生的边界处被划分为Stage，然后DAGScheduler按拓扑顺序运行这些阶段。

每个 Stage 可以是 ShuffleMapStage ，在这种情况下，其Task的结果被输入到其他Stage，
也可以是ResultStage，在该情况下，它的任务通过在RDD上运行函数直接计算Spark动作
（例如count()、save()等）。对于 ShuffleMapStage ，还跟踪每个输出分区所在的节点。

1、以Stage为单位一层一层递归寻找

private[spark] class DAGScheduler(......)extends Logging {private[scheduler] def handleJobSubmitted(jobId: Int,finalRDD: RDD[_],func: (TaskContext, Iterator[_]) => _,partitions: Array[Int],callSite: CallSite,listener: JobListener,properties: Properties): Unit = {var finalStage: ResultStage = nulltry {//创建第一个 Stage finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)} catch {.......}//作业已提交，内部数据清理。barrierJobIdToNumTasksCheckFailures.remove(jobId)val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)clearCacheLocs()//提交finalStage 但首先要递归提交该阶段依赖的所有父stagesubmitStage(finalStage)}//递归方法private def submitStage(stage: Stage): Unit = {val jobId = activeJobForStage(stage)if (jobId.isDefined) {if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {//将 finalStage 传入 获得其 依赖的 Stageval missing = getMissingParentStages(stage).sortBy(_.id)//如果该Stage上一层没有Stage了说明到头了，开始提交这个最初的第一个stageif (missing.isEmpty) {如果上一层没有Stage了那么就提交这层起始的stagesubmitMissingTasks(stage, jobId.get)} else {for (parent <- missing) {//如果Stage前面还有Stage，就继续提交，看是否还有上一层StagesubmitStage(parent)}waitingStages += stage}}} else {abortStage(stage, "No active job for stage " + stage.id, None)}}private def getMissingParentStages(stage: Stage): List[Stage] = {val missing = new HashSet[Stage]  //缺失的Stageval visited = new HashSet[RDD[_]] //访问过的RDD// 我们在这里手动维护一个堆栈，以防止递归访问引起的StackOverflowError//因为递归需要一直将方法压栈 ，最后回归时再弹栈，如果递归太深很容易导致栈溢出val waitingForVisit = new ListBuffer[RDD[_]]waitingForVisit += stage.rdd //等待访问的RDD，最开始就是一个Job的最后那个RDDdef visit(rdd: RDD[_]): Unit = {if (!visited(rdd)) {visited += rdd //这个rdd 已经访问过了val rddHasUncachedPartitions = getCacheLocs(rdd).contains(Nil)if (rddHasUncachedPartitions) {//从这个循环我们可以看出，rdd的依赖是多个的，也就是一个rdd可能来自于1-多个rdd的数据//我们详细看下rdd的依赖关系for (dep <- rdd.dependencies) {dep match {case shufDep: ShuffleDependency[_, _, _] =>//如果RDD和父RDD是Shuffle依赖，就创建一个ShuffleMapStageval mapStage = getOrCreateShuffleMapStage(shufDep, stage.firstJobId)// 只有在基于推送的混洗合并完成后，才将mapStage标记为可用于混洗输出。// 如果没有，后续ShuffleMapStage将不会从合并的输出中读取，因为MergeStatuses不可用。if (!mapStage.isAvailable || !mapStage.shuffleDep.shuffleMergeFinalized) {//添加到在丢失的Stage Set中missing += mapStage} else {//如果跳过并首次被访问，则转发nextAttemptId。否则一旦重试，//  1） 阶段信息中的内容变得扭曲，例如任务编号、输入字节、e.t.c//  2） 第一次尝试从0-idx开始，不会标记为重试mapStage.increaseAttemptIdOnFirstSkip()}case narrowDep: NarrowDependency[_] =>//当这个RDD和它的父RDD时窄依赖时，放到栈中，继续往下寻找waitingForVisit.prepend(narrowDep.rdd)}}}}}//循环这个栈，看是否还有RDD，如果还有就说明还没有窄依赖的父RDDwhile (waitingForVisit.nonEmpty) {visit(waitingForVisit.remove(0))}//返回该stage上一层的stage列表missing.toList}}

2、获取与父RDDs的依赖关系

abstract class RDD[T: ClassTag](@transient private var _sc: SparkContext,@transient private var deps: Seq[Dependency[_]]) extends Serializable with Logging {//我们的依赖关系和分区将通过调用下面的子类方法获得，并在我们被检查点时被覆盖@volatile private var dependencies_ : Seq[Dependency[_]] = _//获取此RDD的依赖关系列表，同时考虑RDD是否为检查点。final def dependencies: Seq[Dependency[_]] = {//OneToOneDependency 是表示父RDD和子RDD的分区之间的一对一依赖关系。checkpointRDD.map(r => List(new OneToOneDependency(r))).getOrElse {if (dependencies_ == null) {stateLock.synchronized {if (dependencies_ == null) {//在此处才真正去获取此RDD的依赖关系dependencies_ = getDependencies}}}dependencies_}}//由子类实现，以返回此RDD如何依赖父RDD。此方法只会被调用一次，因此在其中实现耗时的计算是安全的。//因此不同形态的rdd 获取依赖关系的方法也不同，它的子类有//CoGroupedRDD 有OneToOneDependency (当父子RDD分区数一样时，也就是可以认为调整分区数来调整某些场景的宽窄依赖) 也有 ShuffleDependency//CoalescedRDD 只有 NarrowDependency//ShuffledRDD  只有ShuffleDependency，需要序列化管理器//SubtractedRDD 有OneToOneDependency (当父子RDD分区数一样时) 也有 ShuffleDependency//CartesianRDD  只有固定的两个NarrowDependency//UnionRDD RangeDependency 它 继承了 NarrowDependency 时窄依赖 （表示父RDD和子RDD中分区范围之间的一对一依赖关系）//CoGroupedRDD 、SubtractedRDD  宽窄依赖都有可能，可以通过父子分区数调节//CoalescedRDD 、CartesianRDD  、UnionRDD 都是窄依赖关系 //ShuffledRDD  必然是宽依赖，因为都开始准备序列化管理器拉数据了protected def getDependencies: Seq[Dependency[_]] = deps}//继承了一个NarrowDependency 有一个未实现的方法 即： 获取父rdd的分区数量
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {override def getParents(partitionId: Int): List[Int] = List(partitionId)
}

我们根据一个例子来捋一下

/*** 该程序只作为学习用，没有任何业务知识哈*/
object StageDivision {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("StageDivision").setMaster("local")val sc = new SparkContext(conf)val sourceRdd1 = sc.textFile("file/word_count_data.txt")val sourceRdd2 = sc.textFile("file/word_count_data2.txt")val sourceRdd3 = sc.textFile("file/word_count_data3.txt")val stepOneRdd = sourceRdd1.filter(_.length>10).flatMap(_.split(",")).map(x=>{(x,1)}).groupByKey().map(data=>{(data._1,data._2.filter(_.>(5)).toList.length)})val stepTwoRdd = sourceRdd2.flatMap(_.split(",")).map(x=>{(x,10)})val stepThreeRdd = sourceRdd3.flatMap(_.split(",")).map(x=>{(x,20)}).reduceByKey(_+_)val unionStepOneTwoRdd = stepOneRdd.union(stepTwoRdd)val countRdd = unionStepOneTwoRdd.reduceByKey(_+_).union(stepThreeRdd)val resultMap = countRdd.collect().toMapsc.stop()}

该程序的DAG如下图：

1、submitStage(Stage3)

2、根据RDD依赖关系找上一层Stage：Stage2和Stage1

3、执行submitStage(Stage2)和submitStage(Stage1)

4、根据RDD依赖关系找Stage2和Stage1的上一层Stage,

Stage2没有上一层了，执行submitMissingTasks(stage2, jobId.get)

Stage1还有上一层Stage:执行submitStage(Stage0)

5、根据RDD依赖关系找Stage0的上一层Stage

Stage0没有上一层了，执行submitMissingTasks(stage0, jobId.get)

注意：其中只有Stage3为ResultStage，其他Stage均为ShuffleMapStage

四、提交Task

当根据RDD的依赖关系划分完Stage，就开始从每个分支的最外层Stage提交Task了，即执行submitMissingTasks(最外层Stage)

private[spark] class DAGScheduler(...){//决定任务是否可以将输出提交到HDFS。使用“第一提交者获胜”策略//OutputCommitCoordinator在 driver 和 executor 中都被实例化。在executor 上，它配置了对driver OutputCommitCoordinatorEndpoint的引用，因此提交输出的请求将被转发到driver的OutputCommitCoordinator。private [scheduler] val outputCommitCoordinator = env.outputCommitCoordinatorprivate def submitMissingTasks(stage: Stage, jobId: Int): Unit = {//计算出分区ID的索引Seq。val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()//当一个Stage 启动会有DAGScheduler调度，并对Stage进行初始化stage match {case s: ShuffleMapStage =>outputCommitCoordinator.stageStart(stage = s.id, maxPartitionId = s.numPartitions - 1)case s: ResultStage =>outputCommitCoordinator.stageStart(stage = s.id, maxPartitionId = s.rdd.partitions.length - 1)//计算每一个分区的所有位置val taskIdToLocations: Map[Int, Seq[TaskLocation]] = try {stage match {case s: ShuffleMapStage =>partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMapcase s: ResultStage =>partitionsToCompute.map { id =>val p = s.partitions(id)(id, getPreferredLocs(stage.rdd, p))}.toMap}}//为此 Stage 创建新的尝试 ，也就是Stage有失败重试的机制stage.makeNewStageAttempt(partitionsToCompute.size, taskIdToLocations.values.toSeq)//为 Task 广播二进制文件，用于将 Task 分派给 executor。//请注意，我们广播了RDD的序列化副本，对于每个Task，我们将对其进行反序列化，//这意味着每个任务都会得到一个不同的RDD副本。这在可能修改闭包中引用的对象状态的任务之间提供了更强的隔离。//这在Hadoop中是必要的，因为JobConf/Configuration对象不是线程安全的。var taskBinary: Broadcast[Array[Byte]] = nullvar partitions: Array[Partition] = nulltry {//对于ShuffleMapTask 序列化并广播  (rdd, shuffleDep)//对于ResultTask 序列化并广播 (rdd, func)var taskBinaryBytes: Array[Byte] = nullRDDCheckpointData.synchronized {taskBinaryBytes = stage match {case stage: ShuffleMapStage =>JavaUtils.bufferToArray(closureSerializer.serialize((stage.rdd, stage.shuffleDep): AnyRef))case stage: ResultStage =>JavaUtils.bufferToArray(closureSerializer.serialize((stage.rdd, stage.func): AnyRef))}partitions = stage.rdd.partitions}//如果Task序列化后的大小 > 1 M 就会发出警告if (taskBinaryBytes.length > TaskSetManager.TASK_SIZE_TO_WARN_KIB * 1024) {logWarning(s"Broadcasting large task binary with size " +s"${Utils.bytesToString(taskBinaryBytes.length)}")}//将序列化后的Task广播到每个Executor//有多分区就有多少Task，虽然每个Task会按照分区位置移动到最佳的Executor,但是它们的计算逻辑是一样的，因此可以直接广播taskBinary = sc.broadcast(taskBinaryBytes)}//ShuffleMapStage  ---对应---> ShuffleMapTask//ResultStage   ----对应--->  ResultTaskval tasks: Seq[Task[_]] = try {val serializedTaskMetrics = closureSerializer.serialize(stage.latestInfo.taskMetrics).array()stage match {case stage: ShuffleMapStage =>partitionsToCompute.map { id =>new ShuffleMapTask(...)}case stage: ResultStage =>partitionsToCompute.map { id =>new ResultTask(...)}}}//提交任务//会调用 TaskSchedulerImpl.submitTasks() 是在构建SparkContext时就指定好的taskScheduler.submitTasks(new TaskSet(...))}//获取与特定RDD的分区关联的局部性信息//该方法时一个递归方法，知道找到这个Stage的最除的那个rdd，然后获取每个分区的最佳位置//根据不同rdd的类型有不同的算法，我们单拿出来分析下private[spark]def getPreferredLocs(rdd: RDD[_], partition: Int): Seq[TaskLocation] = {getPreferredLocsInternal(rdd, partition, new HashSet)}private def getPreferredLocsInternal(rdd: RDD[_],partition: Int,visited: HashSet[(RDD[_], Int)]): Seq[TaskLocation] = {//如果分区已被访问，则无需重新访问if (!visited.add((rdd, partition))) {return Nil}//如果分区已经缓存了，那么返回缓存的地址//如果你的rdd会被多个rdd使用，那么可以缓存起来，其他rdd使用时可以直接从缓存拿数据val cached = getCacheLocs(rdd)(partition)if (cached.nonEmpty) {return cached}// 获取分区的首选位置，同时考虑RDD是否为检查点。// 如果rdd设置了首选位置，那么直接使用它val rddPrefs = rdd.preferredLocations(rdd.partitions(partition)).toListif (rddPrefs.nonEmpty) {return rddPrefs.filter(_ != null).map(TaskLocation(_))}//如果RDD具有窄依赖关系，请选择具有任何放置偏好的第一个窄依赖关系的第一个分区。//理想情况下，我们会根据传输大小进行选择，但现在就可以了。//递归方法：同一个Stage的分区也都是窄依赖的，因此需要获取到这个Stage的第一个RDD的那个分区中的最佳位置rdd.dependencies.foreach {case n: NarrowDependency[_] =>for (inPart <- n.getParents(partition)) {val locs = getPreferredLocsInternal(n.rdd, inPart, visited)if (locs != Nil) {return locs}}case _ =>}Nil}}

Task判断分发到哪个Executor

从上面我们已经知道Task的个数 = 这个Stage的最后的那个RDD的分区数量，且需要对每个分区进行递归找到这个Stage最初的那个RDD，找到其数据的存放位置，如果这个位置所在节点也正好有启动的存活的Executor，那么这就是Task要分发的目的地了，下面我们详细看下不同RDD类型对应的位置计算逻辑

1、HadoopRDD

  private[spark] def convertSplitLocationInfo(infos: Array[SplitLocationInfo]): Option[Seq[String]] = {Option(infos).map(_.flatMap { loc =>val locationStr = loc.getLocationif (locationStr != null && locationStr != "localhost") {//优先取在内存中存放的数据节点上的Excutorif (loc.isInMemory) {logDebug(s"Partition $locationStr is cached by Hadoop.")Some(HDFSCacheTaskLocation(locationStr).toString)} else {Some(HostTaskLocation(locationStr).toString)}} else {None}})}

2、ShuffledRDD

  //返回在给定Shuffle中运行给定map输出分区的首选主机，即该分区输出最多的节点。//如果映射输出是预合并的，那么如果合并率高于阈值，则返回合并块所在的节点。def getPreferredLocationsForShuffle(dep: ShuffleDependency[_, _, _], partitionId: Int): Seq[String] = {val shuffleStatus = shuffleStatuses.get(dep.shuffleId).orNullif (shuffleStatus != null) {//检查map输出是否已预合并，合并比率是否高于阈值。如果是这样，合并块的位置是首选位置//判断是否启用基于推送的Shuffle//满足以下条件://    提交应用程序以在YARN模式下运行//    已启用外部洗牌服务//    IO加密已禁用//    序列化器（如KryoSerialer）支持重新定位序列化对象val preferredLoc = if (pushBasedShuffleEnabled) {shuffleStatus.withMergeStatuses { statuses =>val status = statuses(partitionId)val numMaps = dep.rdd.partitions.lengthif (status != null && status.getNumMissingMapOutputs(numMaps).toDouble / numMaps<= (1 - REDUCER_PREF_LOCS_FRACTION)) {Seq(status.location.host)} else {Nil}}} else {Nil}if (preferredLoc.nonEmpty) {preferredLoc} else {//是否计算reduce任务的局部偏好 spark.shuffle.reduceLocality.enabled 默认 true//rdd的分区数 < 1000 (map和reduce任务的数量，超过此数量，我们不会根据map输出大小分配首选位置。我们限制了分配首选位置的作业的大小，因为按大小计算顶部位置变得昂贵。)//也就是如果shuffle过程中分区数大于1000就没有什么最佳位置的概念了，也就造成了速度的不确定性if (shuffleLocalityEnabled && dep.rdd.partitions.length < SHUFFLE_PREF_MAP_THRESHOLD &&dep.partitioner.numPartitions < SHUFFLE_PREF_REDUCE_THRESHOLD) {val blockManagerIds = getLocationsWithLargestOutputs(dep.shuffleId, partitionId,dep.partitioner.numPartitions, REDUCER_PREF_LOCS_FRACTION)if (blockManagerIds.nonEmpty) {blockManagerIds.get.map(_.host)} else {Nil}} else {Nil}}} else {Nil}}