Spark Core解析 2：Scheduler 调度体系

时间  2020-02-13
标签 spark core 解析 scheduler 调度 体系 栏目 Spark 繁體版
原文   原文链接

Spark Core解析 2：Scheduler 调度体系

Overview

调度系统，是贯穿整个Spark应用的主心骨，从调度系统开始入手了解Spark Core，比较容易理清头绪。html

Spark的资源调度采用的是常见的两层调度，底层资源的管理和分配是第一层调度，交给YARN、Mesos或者Spark的Standalone集群处理，Application从第一层调度拿到资源后，还要进行内部的任务和资源调度，将任务和资源进行匹配，这是第二层调度，本文讲的就是这第二层调度。算法

Spark的调度体系涉及的任务包括3个粒度，分别是Job、Stage、Task。
Job表明用户提交的一系列操做的整体，一个具体的计算任务，有明确的输入输出，一个Job由多个Stage组成；
一个Stage表明Job计算流程的一个组成部分，一个阶段，包含多个Task；
一个Task表明对一个分区的数据进行计算的具体任务。shell

层级关系：Job > Stage > Taskapache

在Spark Core 解析：RDD 弹性分布式数据集中，已经解释了RDD之间的依赖，以及如何组成RDD血缘图。segmentfault

因此本文主要目的就是解释清楚：Scheduler将RDD血缘图转变成Stage DAG，而后生成Task，最后提交给Executor去执行的过程。后端

20191212230626.png

Stage

Job的不一样分区的计算一般能够并行，可是有些计算须要将数据进行从新分区，这个过程称做shuffle(混洗)。Shuffle的过程是无法彻底并行的，这时候就会出现task之间的等待，task的数量也可能发生变化，因此Spark中以shuffle为边界，对task进行划分，划分出来的每段称为Stage。app

Stage表明一组能够并行的执行相同计算的task，每一个任务必须有相同的分区规则，这样一个stage中是没有shuffle的。异步

在一个Spark App中，stage有一个全局惟一ID，stage id是自增的。async

20191028171155.png

Stage分为两种：分布式

  • ResultStage：最后执行的stage，负责Job最终的结果输出，每一个Job有且仅有一个ResultStage；
  • ShuffleMapStage：该stage的输出不是最终结果，而是其余stage的输入数据，一般涉及一次shuffle计算。

stage建立流程：

  • 从最终执行action的RDD开始，沿着RDD依赖关系遍历，

一旦发现某个RDD的dependency是ShuffleDependency，就建立一个ShuffleMapStage。

  • 最后建立ResultStage。

example 1

val rg=sc.parallelize(List((1,10),(2,20)))
rg.reduceByKey(_+_).collect

stages-simple.png

这里reduceByKey操做引发了一次shuffle，因此job被切分红了2个stage。

example 2

val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c")))
val rddB=sc.parallelize(List((1,"A"),(2,"B"),(3,"C")))
rddA.join(rddB).collect

stages-join.png

join操做致使rddA和rddB都进行了一次shuffle，因此有3个stage。

example 3

import org.apache.spark.HashPartitioner
val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c"))).partitionBy(new HashPartitioner(3))
val rddB=sc.parallelize(List((1,"A"),(2,"B"),(3,"C")))
rddA.join(rddB).collect

stages-co-join.png

WHAT ?

由于rddA已经定义了Partitioner，这里join操做会保留rddA的分区方式，因此对rddA的依赖是OneToOneDepenency，而对于rddB则是ShuffleDependency。

stage-example-3-2.png

探索：一个RDD被依赖屡次，会如何

val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c")))
rddA join rddA collect

rdd use twice.png

rdd-used-twice.png

一个RDD被两个stage使用了。

小结

综上，stage的划分必定是依据shuffle即ShuffleDependency，跟算子和RDD变量的定义没有很强的关系，example2和3中的join操做rddA.join(rddB).collect看起来如出一辙，但实际产生的stage划分却差异很大。

Task

与stage对应，task也分为两种：

  • ShuffleMapTask：即ShuffleMapStage中的task，主要完成map、shuffle计算。
  • ResultTask：ResultStage中的task，主要完成最终结果输出或者返回结果给driver的任务。

一个stage有多少个partition就会建立多少个task，好比一个ShuffleMapStage有10个partition，那么就会建立10个ShuffleMapTask。

一个Stage中的全部task组成一个TaskSet。

Job Submit

graph TB
R(RDD.action)-->S(SparkContext.runJob)-- RDD -->D(DAGScheduler.runJob)
-- TaskSet -->T(TaskScheduler.submitTasks)-- TaskDescription -->E(Executor.launchTask)

RDD在action操做中经过SparkContext.runJob方法触发Job执行流程，该方法将调用DagScheduler.runJob方法，将RDD传入DagScheduler。而后，DAGScheduler建立TaskSet提交给TaskScheduler，TaskScheduler再将TaskSet封装成TaskDescription发送给Executor，最后Executor会将TaskDescription提交给线程池来运行。

Stage Scheduler(high-level)

DagScheduler

Stage级别的调度是DagScheduler负责的，也是Spark调度体系的核心。

DagScheduler的工做模式

sequenceDiagram
    participant M as main thread
    participant L as eventProcessLoop
    participant E as event thread
    M-->>L: post event
    E-->>L: handle event

DagScheduler内部维护了一个事件消息总线eventProcessLoop(类型为DAGSchedulerEventProcessLoop)，其实就是一个用来存储DAGSchedulerEvent类型数据的队列。

当DagScheduler的一些方法被调用的时候（如submitJob方法），并不会在主线程中处理该任务，而是post一个event(如JobSubmitted)到eventProcessLoop。eventProcessLoop中有一个守护线程，会不断的依次从队列中取出event，而后调用对应的handle(如handleJobSubmitted)方法来执行具体的任务。

Stage调度流程

  • 1.submit job

    DagScheduler.runJob方法会调用submitJob方法，向eventProcessLoop发送一个JobSubmitted类型的消息，其中包含了RDD等信息。当eventProcessLoop接收到JobSubmitted类型的消息，会调用DagScheduler.handleJobSubmitted方法来处理消息。

sequenceDiagram
    participant M as main thread(runJob)
    participant L as eventProcessLoop
    participant E as event thread(handleJobSubmitted)
    M-->>L: post JobSubmitted event
    E-->>L: handle JobSubmitted event

  • 2.create stage

    • DagScheduler在它的handleJobSubmitted方法中开始建立ResultStage。ResultStage中包含了最终执行action的finalRDD，以及计算函数func。
    • ResultStage有个parents属性，这个属性是个列表，也就是说能够有多个parent stage。建立ResultStage时须要先建立它的parent stage来填充这个属性，也就是说要建立ResultStage直接依赖的全部ShuffleMapStage。
    • 经过stage.rdd.dependencies属性，采用宽度优先遍历，一旦发现某个RDD(假设叫rddA)的dependency是ShuffleDependency，就建立一个ShuffleMapStage，ShuffleMapStage中包含的关键信息与ResultStage不一样，是rddA的ShuffleDependency和rddA的ShuffleDependency.rdd，也就是说新建立的ShuffleMapStage持有的信息是他自身的最后一个RDD和该RDD的子RDD的dependency。

    • 建立一个ShuffleMapStage的过程同理会须要建立它的parent stage，也是若干ShuffleMapStage。如此递归下去，直到建立完全部的ShuffleMapStage，最后才完成ResultStage的建立。最后建立出来的这些Stage(若干ShuffleMapStage加一个ResultStage)，经过parent属性串起来，就像这样

      graph TD
A[ResultStage]-- parent -->B[ShuffleMapStage 1]
A-- parent -->C[ShuffleMapStage 2]
B-- parent -->D[ShuffleMapStage 3]

      这就生成了所谓的DAG图，可是这个图的指向跟执行顺序是反过来的，若是按执行顺序来画DAG图，就是常见的形式了：

      graph TD
D[ShuffleMapStage 3]-->C[ShuffleMapStage 2]
C[ShuffleMapStage 2]-->A[ResultStage]
B[ShuffleMapStage 1]-->A[ResultStage]
  • 3.submit stage

    DagScheduler.handleJobSubmitted方法建立好ResultStage后会提交这个stage(submitStage方法)，在提交一个stage的时候，会要先提交它的parent stage,也是经过递归的形式，直到一个stage的全部parent stage都被提交了，它本身才能被提交，若是一个stage的parent尚未完成，则会把这个stage加入waitingStages。也就是说，DAG图中前面的stage会被先提交。当一个stage的parent都准备好了，也就是执行完了，它才会进入submitMissingTasks的环节。

  • 4.submit task

    Task是在DagScheduler（不是TaskScheduler）的submitMissingTasks方法中建立的，包括ShuffleMapTask和ResultTask，与Stage对应。归属于同一个stage的这批Task组成一个TaskSet集合，最后提交给TaskScheduler的就是这个TaskSet集合。

20191029095005.png

Task Scheduler(low-level)

Task的调度工做是由TaskScheduler与SchedulerBackend紧密合做，共同完成的。

TaskScheduler是task级别的调度器，主要做用是管理task的调度和提交，是Spark底层的调度器。

SchedulerBackend是TaskScheduler的后端服务，有独立的线程，全部的Executor都会注册到SchedulerBackend，主要做用是进行资源分配、将task分配给executor等。

Task调度流程

spark task scheduler.png

第一个线程是DAGScheduler的事件处理线程，在其中，Task先通过DAGScheduler（蓝色箭头表示）封装成TaskSet，再由TaskScheduler（绿色箭头）封装成TaskSetManager，并加入调度队列中。

SchedulerBackend在收到ReviveOffers消息时，会从线程池取一个线程进行makeOffers操做，WorkerOffer建立后传递给TaskScheduler进行分配。

图中第二个线程就是SchedulerBackend的一个事件分发线程，从Pool中取出最优先的TaskSetManager，而后将WorkerOffer与其中的Task进行配对，生成TaskDescription，发送给WorkerOffer指定的Executor去执行。

工做流程

TaskScheduler.png

  • 1 DAGScheduler(submitMissingTasks方法中)调用TaskScheduler.submitTasks()建立并提交TaskSet给TaskScheduler；
  • 2 TaskScheduler拿到TaskSet后会建立一个TaskSetManager来管理它，而且把TaskSetManager添加到rootPool调度池中；
  • 3 调用SchedulerBackend.reviveOffers()方法；
  • 4 SchedulerBackend发送ReviveOffers消息给DriverEndpoint；
  • 5 DriverEndpoint收到ReviveOffers消息后，会调用makeOffers()方法建立WorkerOffer，并经过TaskScheduler.resourceOffers()返回offer；
  • 6 TaskScheduler从rootPool获取按调度算法排序后的TaskSetManager列表，取第一个TaskSetManager，逐个给TaskSet的Task分配WorkerOffer，生成TaskDescription(包含offer信息)；
  • 7 调用SchedulerBackend.DriverEndpoint的launchTasks方法，将TaskDescription序列化并封装在LaunchTask消息中，发送给offer指定的executor。LaunchTask消息被ExecutorBackend收到后，会将Task信息反序列化，传给Executor.launchTask()，最后使用Executor的线程池中的线程来执行这个Task。

梳理

Stage,TaskSet,TaskSetManager是一一对应的，数量相等，都是只存在driver上的。
Parition,Task,TaskDescription是一一对应，数量相同，Task和TaskDescription是会被发到executor上的。

TaskScheduler的调度池

与DAGScheduler不一样的是TaskScheduler有调度池，有两种调度实体，Pool和TaskSetManager。
与YARN的调度队列相似，采用了层级队列的方式，Pool是TaskSetManager的容器，起到将TaskSetManager分组的做用。

Schedulable

Schedulable是调度实体的基类，有两个子类Pool和TaskSetManager。

要理解调度规则，必须知道下面几个属性：

  • parent：所属调度池，顶层的调度池为root pool；
  • schedulableQueue：包含的调度对象组成的队列；
  • schedulingMode：调度模式，FIFO or FAIR；
  • weight：权重
  • minShare：最小分配额(CPU核数)
  • runningTasks：运行中task数
  • priority：优先级
  • stageId：就是stageId
  • name：名称

Pool和TaskSetManager对于这些属性的取值有所不一样，从而致使了他们的调度行为也不同。

properties Pool TaskSetManager
weight config 1
minShare config 0
priority 0 jobId
stageId -1 stageId
name config TaskSet_{taskSet.id}
runningTasks Pool所含TaskSetManager的runningTasks和 TaskSetManager运行中task数

Pools建立流程

TaskScheduler有个属性schedulingMode，值取决于配置项spark.scheduler.mode，默认为FIFO。这个属性会致使TaskScheduler使用不一样的SchedulableBuilder，即FIFOSchedulableBuilder和FairSchedulableBuilder。

TaskScheduler在初始化的时候，就会建立root pool，根调度池，是全部pool的祖先。
它的属性取值为：

name: "" (空字符串)
schedulingMode: 同TaskScheduler的schedulingMode属性
weight: 0
minShare: 0

注意root pool的调度模式肯定了。

接下来会执行schedulableBuilder.buildPools()方法，

  • 若是是FIFOSchedulableBuilder，则什么都不会发生。

  • 如果FairSchedulableBuilder

    • 1 依据scheduler配置文件(后面会说)，开始建立pool(能够是多个pool，FIFO，FAIR都有可能，取决于配置文件)，并都加入root pool中。
    • 2 若是如今root pool中没有名为"default"的pool(即配置文件中没有定义一个叫default的pool)，建立default pool，并加入root pool中。

这时default pool它的属性取值是固定的：

name: "default"
schedulingMode: FIFO
weight: 1
minShare: 0

Task加入pool流程

当TaskScheduler提交task的时候，会先建立TaskSetManager，而后经过schedulableBuilder添加到pool中。

  • 若是是FIFOSchedulableBuilder，则会直接把TaskSetManager加入root pool队列中。

  • 如果FairSchedulableBuilder

    • 1 从spark.scheduler.pool配置获取pool name，没有定义则用'default'；
    • 2 从root pool遍历找到对应名称的pool，把TaskSetManager加入pool的队列。若是没有找到，则建立一个该名称的pool，采用与default pool相同的属性配置，并加入root pool。

调度池结构

通过上面两部分，最终获得的调度池结构以下：

spark.scheduler.mode=FIFO

20191128210416.png

spark.scheduler.mode=FAIR

20191128210432.png

Fair Scheduler pools配置

Fair Scheduler Pool的划分依赖于配置文件，默认的配置文件为'fairscheduler.xml'，也能够经过配置项"spark.scheduler.allocation.file"指定配置文件。

煮个栗子，文件内容以下：

<?xml version="1.0"?>
<allocations>
  <pool name="prod">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
  <pool name="test">
    <schedulingMode>FIFO</schedulingMode>
    <weight>2</weight>
    <minShare>3</minShare>
  </pool>
</allocations>

这里配置了两个pool，prod和test，而且配置了相关属性，这两个pool都会添加到root pool中。

调度算法

以SchedulingAlgorithm为基类，内置实现的调度算法有两种FIFOSchedulingAlgorithm和FairSchedulingAlgorithm，其逻辑以下：

  • FIFO: 先进先出，优先级比较算法以下，

    • 1.比较priority，小的优先；
    • 2.priority相同则比较StageId，小的优先。

  • FAIR：公平调度，优先级比较算法以下，

    • 1.runningTasks小于minShare的优先级比不小于的优先级要高。
    • 2.若二者运行的runningTasks都比minShare小，则比较minShare使用率(runningTasks/max(minShare,1))，使用率越低优先级越高。
    • 3.若二者的minShare使用率相同，则比较权重使用率(runningTasks/weight)，使用率越低优先级越高。
    • 4.若权重也相同，则比较name，小的优先。
Pool为FIFO模式下的几种情形

TaskSetManager之间的比较，其实就是先比较jobId再比较stageId，谁小谁优先，意味着就是谁先提交谁优先。

Pool之间的比较，不存在！FIFO的pool队列中是不会有pool的。

Pool为FAIR模式下的几种情形

TaskSetManager之间的比较，由于minShare=0，weight=1，FAIR算法变成了：

  • 1 runningTasks小的优先
  • 2 runningTasks相同则比较name

Pool之间的比较，就是标准的FAIR算法。

当root pool为FAIR模式，先取最优先的pool，再从pool中，按pool的调度模式取优先的TaskSetManager。

开始使用FAIR mode

启用FAIR模式：

  • 1 准备好fairscheduler.xml文件
  • 2 启动参数添加 --conf spark.scheduler.mode=FAIR
  • 3 运行启动命令，如spark-shell --master yarn --deploy-mode client --conf spark.scheode=FAIR

ui-fair.png

启动后若是直接运行Job会自动提交到default pool，那么如何提交Job到指定pool？
SparkContext.setLocalProperty("spark.scheduler.pool","poolName")

若是每次只运行一个Job，开启FAIR模式的意义不大，那么如何同时运行多个Job？
要异步提交Job，须要用到RDD的async action，目前有以下几个：

countAsync
collectAsync
takeAsync
foreachAsync
foreachPartitionAsync

举个例子：

sc.setLocalProperty("spark.scheduler.pool","test")
b.foreachAsync(_=>Thread.sleep(100))
sc.setLocalProperty("spark.scheduler.pool","production")
b.foreachAsync(_=>Thread.sleep(100))

这样就会有两个任务在不一样的pool同时运行：

pools.png

FAIR mode应用场景

场景1：Spark SQL thrift server
做用：让离线任务和交互式查询任务分配到不一样的pool，给交互式查询任务更高的优先级，这样长时间运行的离线任务就不会一直占用全部资源，阻塞交互式查询任务。

场景2：Streaming job与Batch job同时运行
做用：好比用Streaming接数据写入HDFS，可能产生不少小文件，能够在低优先级的pool定时运行batch job合并小文件。

另外能够参考Spark Summit 2017的分享：Continuous Application with FAIR Scheduler

参考

Spark内核设计的艺术

spark任务调度FIFO和FAIR的详解

Job Scheduling

转载请注明原文地址：
https://liam-blog.ml/2019/11/07/spark-core-scheduler/

查看更多博主文章

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程