Spark2.2任务提交运行机制及源码解析

云计算 waitig 1023℃ 百度已收录 0评论

源码版本:2.2
参考《Spark内核机制解析及性能调优》
如有错误请指正

一、Spark运行的核心概念

Spark调度器的设计体现得非常简洁清晰和高效,其输入是Spark RDD,输出是Spark执行器(Executor)。正是Spark调度器的设计思想极大地区分出了基于MapReduce模型的Hadoop和基于DAG模型的Spark。
主要内容包括Spark运行核心概念,Spark Driver Program剖析,Spark作业(Job)的触发,高层DAG调度器(DAGScheduler),底层的Task调度器(TaskScheduler)和调度器的通信终端(Scheduler-Backend)。 

1.1 Spark运行的基本对象

RDD是Spark运行的基本对象。
RDD是一种数据模型,与分布式的共享内存类似,RDD具有以下几个特征

  1. 是只读的内存数据,但可以持久化
  2. 是可以分区的数据集合
  3. 是一个对象,可以调用它的方法执行一些变换操作(Transformation),如flapMap、filter等
  4. 是可恢复。与数据多副本的备份方式不同,RDD的恢复可以通过重复执行变换操作(Transformation)得到

有下列两种方式可以创建RDD

  1. 从持久化的数据上创建,如硬盘和HDFS上的文件
  2. 从其他RDD上通过变换Transformation操作创建

1.2 Spark运行框架及各组件的基本运行原理

二、Spark Driver Program剖析

2.1 什么是Spark Driver Program

Driver是运行Application的Main函数,并且新建SparkContext实例的程序。
初始化SparkContext是为了准备Spark应用程序的环境,在Spark中由SparkContext负责与集群进行通信、资源的申请,以及任务的分配和监控等。
当Worker节点中的Executor运行完Task后,Driver同时负责将SparkContext关闭,通常也可以使用SparkContext来代表驱动程序。 

2.2 SparkContext原理剖析

SparkContext是用户通往Spark集群的唯一入口,可以用来在Spark集群中创建RDD、累加器Accumulator和广播变量Braodcast Variable。
SparkContext 也是整个Spark应用程序中至关重要的一个对象,可以说是整个应用运行调度的核心(不是指资源调度)。
SparkContext在实例化的过程中会初始化DAGScheduler、TaskScheduler和SchedulerBackend,而当RDD的action出发了作业Job之后,
SparkContext会调用DAGScheduler将整个Job划分成几个小的阶段(Stage),TaskScheduler会调度每个Stage的任务(Task)应该如何处理。另外,SchedulerBackend管理整个集群中为这个当前的应用分配的计算资源(Executor). 

2.3 SparkContext源码解析 (2.2中被SparkSession取代)

2.3.1构造函数

SparkContext提供了多种不同的构造函数

  1. 默认构造函数, 完全以系统配置作为参数进行初始化,例如,通过./bin/spark-submit提交任务

    2.  提供master url和应用名称进行初始化

    3.  提供master url,appname,sparkhome,jars和环境变量

等等

2.3.2成员变量

成员变量都是私有的,SparkContext另外提供了一系列的方法允许外界进行访问
SparkContext中比较重要的成员:
SchedulerBackend
TaskScheduler
DAGScheduler 

2.3.3初始化和校验

  1. 初始化异步监听bus

监听spark事件,用于SparkUI的跟踪管理

    2.  创建spark环境变量

    3.  启动心跳接收器

在创建taskScheduler之间需要先注册HeartbeatReceiver,因为Executor在创建时回去检索HeartbeatReceiver

    4.  创建SchedulerBackend、TaskScheduler、DAGScheduler

    5.  启动TaskScheduler

在TaskScheduler被DAGScheduler引用后,就可以进行启动

    6. pos init

2.3.4方法相关

  1. RDD相关
  2. Accumulator相关
  3. Job相关
  4. Cluster相关
  5. 缓存相关

三、Spark Job的触发

3.1 Job的执行逻辑

典型的Job逻辑执行有下列四个步骤

  1. 从数据源(本地File,内存数据结构,HDFS,HBase等)读取数据,创建最初的RDD
  2. 对RDD进行一系列的transformation()操作,每一个transformat会产生一个或多个包含不同类型T的RDD[T]

T可以是Scala里的基本类型或数据结构,不限于(K,V)。但如果是(K,V),K不能是Array等复杂类型(因为难以在复杂类型上定义partition函数)

  1. 队最后的final RDD进行action()操作,每个partition计算后产生结果Result
  2. 将Result回送到Driver端,进行最后的f(list[result])计算。

3.2 Job具体的物理执行

Spark Application里可以产生1个或者多个Job,程序中一般而言可以有不同的Action,每一个Action一般也会触发一个Job.
有了Job的逻辑执行图后,如何合理划分Stage,并确定Task的类型和个数?
基于Pipeline的思想:数据用的时候再算,而且数据是流到要计算的位置的。我们可以将整个流程图看成一个Stage,为最后一个finalRDD中的每个partition分配一个Task。
Spark算法构造和物理执行时最基本的核心:最大化Pipeline。
整体流程:整个Computing Chain根据数据依赖关系自后向前建立,遇到ShuffleDependency后形成Stage。在每个Stage中,每个RDD的compute()调用parentRDD.iter()来将parent RDD中的Record一个个fetch过来。
stage内部计算的逻辑是完全一样的,只是计算的数据不同而已。 

3.3 Job触发流程源代码解析

3.3.1 SparkContext提交Job

runJob有多个重载方法,下面是一个最典型的
触发针对一个RDD的所有partition的计算,并将处理结果返回在一个数组里 

调用runJob的重载方法
触发一个在RDD的给定partition上运行的Job,并将处理结果返回在一个数组里 


在这个方法里首先会针对func调用一个clean,目的是针对传输的闭包进行清理,保证func可以正常序列化发送到task上

然后再调用一个重载方法
这里的ctx是从哪里来的???这是什么语法   ctx到底是从哪里来的。。。

重载方法如下

方法里的index是从哪里来的???
构造出result集合,最后一次进入SparkContext 中runJob的另一个同名重载方法,构造出了result handler函数
触发一个Job处理一个RDD的指定部分的partition,并把处理结果给指定的handler函数,这是Spark的所有Action 的主入口 


向高层调度器DAGScheduler提交Job,从而获得执行结果

3.3.2 DAGScheduler接收Job

3.3.2.1 DAGScheduler实例化

DAGScheduler是在SparkContext中实例化的,相关实例化代码如下

3.3.2.2 DAGScheduler提交Job

RDD的action操作会调用SparkContext中的runJob方法,SparkContex的runJob方法最终会调用自身维护的已经实例化过的DAGScheduler的runJob

DAGScheduler的runJob方法,会先调用submitJob方法构造一个JobWaiter对象(干嘛的?就是一个future),在submitJob方法,真正执行提交job操作

  1. 先校验partition是否有不合理partition
  2. 获取递增的jobId
  3. 构造新的JobWaiter对象,JobWaiter对象等待DAGScheduler 任务完成,任务结束时,它会将task的结果传递给result handler
  4. 提交一个JobSubmitted任务给eventProcessLoop

eventProcessLoop是一个DAGSchedulerEventProcessLoop,它的父类是EventLoop
EventLoop接收事件,放在队列中,不断从队列中获取新事件,进行处理

DAGSchedulerEventProcessLoop重写了onReceive,定义了对于不同事件的不同处理形式

不同Job的处理方式如下:

  1. JobSubmitted =》 handleJobSubmitted
  2. MapStageSubmitted =》handleMapStageSubmitted
  3. StageCancelled =》 handleStageCancellation
  4. JobCancelled =-》 handleJobCancellation
  5. JobGroupCancelled =》handleJobGroupCancelled
  6. AllJobsCancelled =》 doCancelAllJobs
  7. ExecutorAdded =》 handleExecutorAdded
  8. ExecutorLost =》handleExecutorLost
  9. BeginEvent =》 handleBeginEvent
  10. GettingResultEvent =》 handleGetTaskResult
  11. CompletionEvent =》handleTaskCompletion
  12. TaskSetFailed =》handleTaskSetFailed
  13. ResubmitFailedStages =》resubmitFailedStages

可以看出,这里不仅是处理RDD的action操作,还处理了所有的执行流程相关事件
 

3.3.2.3 DAGScheduler划分Stage

这里主要分析一下JobSubmitted事件 handleJobSubmitted

处理步骤:

  1. 创建finalStage,获取所有依赖的Parent Stage列表,以及ResultStage的Stage ID

    2. 创建Job

    3. 将job设置到stage中

    4. 设置监听事件,提交stage


提交stage,根据RDD的依赖关系,递归提交所有的Stage,实际提交Stage时,调用的是submitMissingTasks

    5. 提交Task  submitMissingTasks
submitMissingTasks 方法中体现了利用RDD的本地性来得到Task的本地性,从而获取Stage内部Task的最佳位置。通过调用getPreferredLocs方法得到Task的最佳位置,并把结果存放到taskIdToLocations Map中方便调用。



submitMissingTasks中调用了getPreferredLocs方法来获取数据本地化信息

调用方法getPreferredLocsInternal

调用方法getPreferredLocsInternal

处理流程

  1. 检查数据是否在之前的迭代中已经被访问过了,没有必要重复访问

      2.  检查数据是否cache,如果有,直接返回

      3.  检查内存数据结构中是否存在当前partition的数据本地信息,如果有,直接返回

      4.  如果RDD有窄依赖,判断RDD本身是否有首选地址,如有,就用来构造TaskLocation并返回

3.3.3 TaskScheduler接收Task

3.3.3.1 TaskScheduler实例化

TaskScheduler的实例化发生在SparkContext实例化过程中

具体实例化时会根据当前启动模式进行选择如何初始化TaskSchedulerImpl和SchedulerBackend

3.3.3.2 TaskScheduler初始化

    TaskScheduler的初始化过程其实就是在实现类TaskSchedulerImpl的initialize方法中,初始化过程中会设置对SchedulerBackend对象的引用,实例化SchedulerBuilder具体实现类的对象用来创建和管理TaskSetManager池。

TaskSchedulerImpl中默认的调度模式是FIFO

初始化过程initailize会设置对SchedulerBackend对象的引用,根据调度模式,初始化ScheduleBuiler

然后开始构建不同SchedulableBuilder的Pool 
FIFO的buildPools实现为空
FAIR的buildPools实现如下,啊?看不懂目的

3.3.3.3 TaskScheduler启动

TaskScheduler实例对象在DAGScheduler实例化后启动,并且TaskScheduler启动的过程由TaskSchedulerImpl具体实现。在启动过程中,主要是调用SchedulerBackend的启动方法,然后对不是本地部署模式并且开启任务的推测执行???(设置spark.speculation为true)情况,根据配置判断是否周期性地调用TaskSetManager的checkSpeculatable方法检测任务的推测执行???、 
 
    SparkDeploySchedulerBackend的start方法最终会注册应用程序AppClient。 
创建完Scheduler之后执行TaskScheduler的启动

TaskSchedulerImpl中的start方法直接调用了backend.start,但只有在standalone模式下才会调用

3.3.3.4 TaskScheduler提交任务

 TaskSchedulerImpl启动后,就可以接收DAGScheduler的submitMissingTasks方法提交过来的TaskSet进行进一步处理了。
    对于ShuffleMapStage类型的Stage,DAGScheduler初始化一组ShuffleMapTask实例对象;对于ResultStage类型的Stage,DAGScheduler初始化一组ResultTask实例对象。   怎么区分的stage类型???
    最后,DAGScheduler将这组ResultTask实例对象封装成TaskSet实例对象提交给TaskSchedulerImpl
    注意,ShuffleMapTask是根据Stage所依赖的RDD的partition分布产生跟partition数量相等的Task,这些Task根据partition的本地性分布在不同的集群节点;ResultTask负责输出整个Job的结果。
DAGScheduler.submitMissingTasks流程

  1. 区分stage类型ShuffleMapStage,生成新的StageStatus
  2. 区分stage类型,查看数据资源位置信息
  3. 区分stage类型,将task进行序列化,广播出去
  4. 区分stage类型,创建一组ShuffleMapTask/ ResultTask实例对象

TaskSchedulerImpl.submitTasks
TaskSchedulerImpl在submitTasks中初始化一个TaskSetManager,并通过SchedulerBuilder对其进行生命周期管理,最后调用SchedulerBackend的reviveOffers方法进行TaskSet所需资源的分配。在TaskSet得到足够的资源后,在SchedulerBackend的launchTasks方法中将TaskSet中的Task一个个地发送到Executor中去执行。 

 

四、高层的DAGScheduler

4.1 DAGScheduler的定义

2.2章节中介绍了RDD DAG构建了RDD的数据流,即RDD的各个分区数据是从哪里来的,还进一步分析了RDD DAG的生成机制和RDD DAG的逻辑视图。
RDD DAG还构建了基于数据流之上的操作算子流,即RDD的各个分区的数据总共会经过哪些Transformation和Action这两种类型的一系列操作的调度运行,从而RDD先被Transformation操作转换为新的RDD,然后被Action操作将结果反馈到Driver Program或存储到外部存储系统上。
    上面提到的一系列操作的调度运行其实是DAG提交给DAGScheduler来解析完成的,DAGScheduler是面向Stage的高层级的调度器,DAGScheduler把DAG拆分成很多Tasks,每组Tasks都是一个Stage。解析时是以Shuffle为边界反向解析构建Stage,每当遇到Shuffle就会产生新的Stage,然后以一个个TaskSet的形式提交给底层调度器TaskScheduler。另外,DAGScheduler需要记录哪些RDD被存入磁盘等物化动作,同时要寻求Task的最优化调度,如在Stage内部数据的本地性等。DAGScheduler还需要监视因为Shuffle跨节点输出可能导致的失败,如果发现这个Stage失败,可能就要重新提交该Stage。
    概念说明:

  1. Stage:一个Job需要拆分成多组任务来完成,每组任务由Stage封装。跟一个Job的所有涉及的partitionRDD类似,Stage之间也有依赖关系
  2. TaskSet:一组任务就是一个TaskSet,对应一个Stage,其中,一个TaskSet的所有Task之间没有Shuffle依赖,因此互相之间可以并行运行
  3. Task:一个独立的工作单元,由Driver Program发送到Executor上去执行,通常情况下,一个Task处理RDD的一个Partition的数据。根据Task返回类型的不同,Task又分为ShuffleMapTask和ResultTask。

4.2 DAGScheduler的实例化

参考

4.3 DAGScheduler划分Stage的原理

从后往前,根据是否需要Shuffle来划分Stage

4.4 DAGScheduler划分Stage的具体算法

RDD的Action操作算子如count会触发一个Spark Job,其实是RDD的count方法调用了SparkContext的runJob方法。然后,在SparkContext的runJob方法中调用3次重载的runJob方法。最后被调用的重载runJob方法中调用了DAGScheduler的runJob方法,从而进入DAGScheduler的源代码。
    在DAGScheduler的源代码中,DAGScheduler的runJob方法中调用submitJob方法。submitJob方法中最重要的是创建一个JobWaiter对象,以及创建JobSubmitted事件对象把JobWaiter对象发送给DAGScheduler的内嵌类DAGSchedulerEventProcessLoop对象实例,DAGSchedulerEventProcessLoop在doOnReceive方法中反过来调用了DAGScheduler中实现划分Stage算法很关键的handleJobSubmitted方法。
DAGScheduler的handleJobSubmitted方法中调用了newResultStage方法,newResultStage方法根据finalRDD创建finalStage,这时便真正开始了Stage的划分。DAGScheduler的handleJobSubmitted方法中最后调用submitStage方法,根据RDD的依赖的关系,递归提交所有的Stage。
    DAGScheduler的getShuffleDependencies方法是划分Stage的核心实现。在该方法中,Stage的划分由最后一个Rsult Stage开始,从后往前回溯边划分边创建。给定的RDD会先被压入Stack,以便被逐个访问。对每个被访问的RDD,判断其Dependency是否是ShuffleDependency宽依赖,如果是,就调用getShuffleMapStage方法获取或创建Stage,并放到一个临时的HashSet中,便于统一最后转为列表返回;如果不是,就把Dependency的RDD也压入Stack中,继续判断该RDD的Denpendency,该方法的结束标志是Stack中所有压入的临时RDD都被访问过。
    Stage划分完成后,DAGScheduler就会回到HandleJobSubmitted方法中调用submitStage方法。在submitStage方法中,从finalStage开始回溯,直到没有Parent Stage为止,提交整个Job的所有Stage,在某一个Stage,DAGScheduler会调用submitMissingTasks方法把Tasks提交给TaskScheduler进行细粒度的Task调度。
JobWaiter    JobSubmitted  DAGSchedulerEventProcessLoop   handleJobSubmitted 

4.5 Stage内部Task获取最佳位置的算法

五、底层的Task调度器TaskScheduler

TaskScheduler的核心任务是提交TaskSet到集群运算并汇报结果。

  1. 为TaskSet创建和维护一个TaskSetManager,并追踪任务的本地性以及错误信息
  2. 遇到Straggle任务会放到其他节点进行重试
  3. 向DAGScheduler汇报执行情况,包括在Shuffle输出丢失时报告fetch failed错误等信息

5.1 TaskScheduler原理剖析

DAGScheduler将划分的一系列Stage(每个Stage封装一个TaskSet),按照Stage的先后顺序依次提交给底层的TaskScheduler去执行。
    DAGScheduler在SparkContext实例化时,TaskScheduler和SchedulerBackend就已经先在SparkContext的createTaskScheduler方法中创建出实例对象了。
    TaskSchedulerImpl在createTaskScheduler方法中实例化后,就立即调用自己的initialize方法把SparkDeploySchedulerBackedn的实例对象传进来,从而赋值给TaskSchedulerImpl的backend。在TaskScheduler的initialize方法中,根据调度模式的配置创建实现了SchedulerBuilder接口的相应实例对象,并且创建的对象会立即调用buildPools创建相应数量的Pool存放和管理TaskManager的实例对象。

  1. FIFOSchedulableBuilder
  2. FairSchedulableBuilder

在createTaskScheduler方法返回后,TaskSchedulerImpl通过DAGScheduler的实例化过程设置DAGScheduler的实例对象,然后调用自己的start方法,在SparkDeploySchedulerBackend的start方法中会最终注册应用程序AppClient。TaskSchedulerImpl的start方法中还会根据配置判断是否周期地检测任务的推测执行。
    TaskScheduler启动后,就可以接收DAGScheduler的submitMissingTasks方法提交过来的TaskSet进行进一步处理。TaskSchedulerImpl在submitTasks中初始化一个TaskSetManager对其进行生命周期管理,当TaskSchedulerImpl得到Worker节点上的Executor计算资源时,会通过TaskManager来发送具体的Task到Executor上执行计算。
    如果Task执行过程中有错误导致失败,会调用TaskSetManager来处理Task失败的情况,进而通知DAGScheduler结束当前的Task。TaskSetManager会将失败的Task再次添加到待执行的Task队列中。Task的默认失败次数是4次。
    如果Task执行完毕,执行的结果会反馈给TaskManager,由TaskManager通知DAGScheduler。DAGScheduler根据是否还存在待执行的Stage,继续迭代提交对应的TaskSet给TaskScheduler去执行,或者输出Job的结果。 

5.2 TaskScheduler源代码解析

。。。

六、调度器的通信终端SchedulerBackend

。。。


本文由【waitig】发表在等英博客
本文固定链接:Spark2.2任务提交运行机制及源码解析
欢迎关注本站官方公众号,每日都有干货分享!
等英博客官方公众号
点赞 (0)分享 (0)