DagScheduler 和 TaskScheduler

DagScheduler 和 TaskScheduler 的任务交接

spark 调度器分为两个部分, 一个是 DagScheduler, 一个是 TaskScheduler, DagScheduler 主要是用来把一个 Job 根据宽依赖划分为多个Stage(阶段),

对于划分出来的每个 stage 都抽象为一个  TaskSet任务集 交给  TaskScheduler 来进行进一步的调度运行,

我们来看一张图, 来理清里面的概念, 我们用户编程使用的 RDD, 每个 RDD都有一个分区数,  这个分区数目创建 RDD 的时候有一个初始值,运行过程中,根据配置的 parallelism 参数 和 shuffle 过程中显示指定的分区数目 来调整个数

我们可以看到, 一个 task 对应一个 stage 里面一个分区数据的处理任务,  task 又分为 ShuffleMapTask 和 ResultTask , 区分任务是中间阶段的任务 还是最后一个阶段的任务。

而一个 stage 里面 所有分区的任务集合 就被包装为一个  TaskSet 交给了 TaskScheduler,

TaskScheduler 调度方式

TaskScheduler 会为每个 TaskSet 创建一个 TaskScheduler, 作为一个调度单位, 放在任务池子里面,

调度池分为两种, 一种使用 FIFO调度方式 , 还有一种使用 Fair调度方式

FIFO调度方式

这种方式 rootPool 下面直接就是 TaskSetManager , 没有子 Pool,

根据  FIFOSchedulingAlgorithm 算法排序, 这种方式排序方式很简单, 直接就是先进先出队列的排序方式对多个  TaskSetManager 进行排队,

Fair调度方式

这种方式 rootPool 是根 pool, 下一级是 用户定义的 Pool,  这一层是为了给不同的用户定义不同的优先级用的,

用户 Pool 下面才是 TaskSetManager

FairSchedulingAlgorithm 算法,排序方式是由两个因子控制,

  • weight: 控制资源池相对其他资源池,可以分配到资源的比例。默认所有资源池的weight都是1。如果你将某个资源池的weight设为2,那么该资源池中的资源将是其他池子的2倍。如果将weight设得很高,如1000,可以实现资源池之间的调度优先级 – 也就是说,weight=1000的资源池总能立即启动其对应的作业。
  • minShare:除了整体weight之外,每个资源池还能指定一个最小资源分配值(CPU个数),管理员可能会需要这个设置。公平调度器总是会尝试优先满足所有活跃(active)资源池的最小资源分配值,然后再根据各个池子的weight来分配剩下的资源。因此,minShare属性能够确保每个资源池都能至少获得一定量的集群资源。minShare的默认值是0。

排序也是递归的, 因为 rootPool 下面有多个 用户自己的 Pool, 要先根据 FairSchedulingAlgorithm 算法对多个 用户的Pool 排序, 然后对一个 Pool 中的多个 TaskSetManager 也使用 FairSchedulingAlgorithm 算法排序。

任务实际运行的触发方式

触发方式有两种,

  • 一堆任务  从 DagScheduler 到 TaskSetManager 提交过来了, 这个时候可能有 大量 executor 的cpu 都闲着呢, 所以要 唤醒他们过来领任务去执行,  这种我们叫做唤醒方式,
  • 一个正在执行的任务跑完了,  executor 报告任务已经执行完的时候, 这个时候 这个 cpu 又闲着了,看看有没有 任务去领一下,  这种我们叫做干完了手里的活接着干方式
  • 有可能增加了新的 Executor  ,这个 executor 来注册了, 相当于产生了新的劳动力, 肯定也要去领活干, 我们叫新劳动力领活干方式,

对于 唤醒方式 就是  TaskScheduler 把  TaskScheduler, 作为一个调度单位, 放在任务池子里面后, 然后 调用 reviveOffers 来唤醒, 里面是调用 makeOffers() 方法,

对于 干完了手里的活接着干方式,  就是 接受到了 StatusUpdate 消息,  会去调用一下  makeOffers(executorId)

对于  新劳动力领活干方式 就是接受到  RegisterExecutor 消息后调用一下  makeOffers()

我们来看下  makeOffers() 方法,

 val workOffers = IndexedSeq(
          new WorkerOffer(executorId, executorData.executorHost, executorData.freeCores))
        launchTasks(scheduler.resourceOffers(workOffers))

把执行单位包装为 一个 WorkerOffer ,然后调用  taskSchedule 的 resourceOffers方法, 这个方法的注释如下

Called by cluster manager to offer resources on slaves. We respond by asking our active task

sets for tasks in order of priority. We fill each node with tasks in a round-robin manner so

that tasks are balanced across the cluster.

这个是  resourceOffers 函数的注释,  就是从把任务池子 中的任务排队, 然后取出最前面的任务, 来和 执行单位相结合, 这里需要注意的是, 会把任务尽可能均匀的分配到每个node 上,  一个任务和一个执行单位的结合包装为一个 TaskDescription,  然后把任务发送到执行单位上去执行,

上图中我们可以看到, 只要有空闲的 executor 就会提供资源给 task, 首先要把 workerOffer shuffle  打乱一下, 免得过分蹂躏个别的 executor。

时间: 2024-08-02 11:22:51

DagScheduler 和 TaskScheduler的相关文章

spark 笔记 7: DAGScheduler

在前面的sparkContex和RDD都可以看到,真正的计算工作都是同过调用DAGScheduler的runjob方法来实现的.这是一个很重要的类.在看这个类实现之前,需要对actor模式有一点了解:http://en.wikipedia.org/wiki/Actor_model http://www.slideshare.net/YungLinHo/introduction-to-actor-model-and-akka 粗略知道actor模式怎么实现就可以了.另外,应该先看看DAG相关的概念

spark1.1.0源码阅读-dagscheduler and stage

1. rdd action ->sparkContext.runJob->dagscheduler.runJob 1 def runJob[T, U: ClassTag]( 2 rdd: RDD[T], 3 func: (TaskContext, Iterator[T]) => U, 4 partitions: Seq[Int], 5 callSite: String, 6 allowLocal: Boolean, 7 resultHandler: (Int, U) => Unit

【Spark】DAGScheduler源码浅析

DAGScheduler DAGScheduler的主要任务是基于Stage构建DAG,决定每个任务的最佳位置 记录哪个RDD或者Stage输出被物化 面向stage的调度层,为job生成以stage组成的DAG,提交TaskSet给TaskScheduler执行 重新提交shuffle输出丢失的stage 每一个Stage内,都是独立的tasks,他们共同执行同一个computefunction,享有相同的shuffledependencies.DAG在切分stage的时候是依照出现shuff

Spark核心作业调度和任务调度之DAGScheduler源码

前言:本文是我学习Spark 源码与内部原理用,同时也希望能给新手一些帮助,入道不深,如有遗漏或错误的,请在原文评论或者发送至我的邮箱 [email protected] 摘要: 1.作业调度核心--DAGScheduler 2.DAGScheduler类说明 2.1DAGScheduler 2.2ActiveJob 2.3Stage 2.4Task 3.工作流程 3.1划分Stage 3.2生成Job,提交Stage 3.3任务集的提交 3.4任务作业完成状态的监控 3.5任务结果的获取 内容

【Spark】DAGScheduler源代码浅析

DAGScheduler DAGScheduler的主要任务是基于Stage构建DAG,决定每个任务的最佳位置 记录哪个RDD或者Stage输出被物化 面向stage的调度层.为job生成以stage组成的DAG.提交TaskSet给TaskScheduler运行 又一次提交shuffle输出丢失的stage 每个Stage内.都是独立的tasks,他们共同运行同一个computefunction,享有同样的shuffledependencies.DAG在切分stage的时候是按照出现shuff

Spark任务流程笔记

Spark学习笔记总结 02. Spark任务流程 1. RDD的依赖关系 RDD和它依赖的父RDD(s)的关系有两种不同的类型,即窄依赖(narrow dependency)和宽依赖(wide dependency) 窄依赖窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用总结:窄依赖我们形象的比喻为独生子女 宽依赖宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition 总结:窄依赖我们形象的比喻为超生(多个子女)查看RDD依

底层战详解使用Java开发Spark程序(DT大数据梦工厂)

Scala开发Spark很多,为什么还要用Java开发原因:1.一般Spark作为数据处理引擎,一般会跟IT其它系统配合,现在业界里面处于霸主地位的是Java,有利于团队的组建,易于移交:2.Scala学习角度讲,比Java难.找Scala的高手比Java难,项目的维护和二次开发比较困难:3.很多人员有Java的基础,确保对Scala不是很熟悉的人可以编写课程中的案例预测:2016年Spark取代Map Reduce,拯救HadoopHadoop+Spark = A winning combat

彻底解密WordCount运行原理(DT大数据梦工厂)

主要内容: 数据流动视角解密WordCount RDD依赖关系视角解密WordCount DAG与Lineage的思考 ==========数据流动视角============ 新建文件,里面输入 Hello Spark Hello Scala Hello Hadoop Hello Flink Spark is awesome 修改代码: package com.dt.spark.SparkApps.cores; import java.util.Arrays; import java.util

[Spark内核] 第35课:打通 Spark 系统运行内幕机制循环流程

本课主题 打通 Spark 系统运行内幕机制循环流程 引言 通过 DAGScheduelr 面向整个 Job,然后划分成不同的 Stage,Stage 是從后往前划分的,执行的时候是從前往后执行的,每个 Stage 内部有一系列任務,前面有分享過,任务是并行计算啦,这是并行计算的逻辑是完全相同的,只不过是处理的数据不同而已,DAGScheduler 会以 TaskSet 的方式把我们一个 DAG 构造的 Stage 中的所有任务提交给底层的调度器 TaskScheduler,TaskSchedu