zoukankan      html  css  js  c++  java
  • Spark源码走读3——Job Runtime

    下面主要说明作业提交的的具体运行环境,这里从SparkContext中的runJob方法开始跟踪它的源码过程。下面的图简要的描述了Job运行的过程


    runJob的源码如下:


    这里主要有三个函数:

    Clean(func):主要是清理关闭一些内容,比如序列化。

    runJob(...):将任务提交给DagScheduler。

    doCheckpoint():保存当前RDD,在Job完成之后调用父rdd。

    这里主要看runJob方法

    DagScheduler

    跟踪源码,进入了DagScheduler的runJob方法。这这里方法中会直接提交Job,然后等待返回结果。成功时JobSucceeded什么都不做,失败则抛出异常。

    进去源码submitJob方法,这方法提交作业到调度器,源码如下:


    跟踪源码:eventProcessActor是DagScheduler的事件驱动也是一个Actor。这里发送了一个消息eventProcessActor !JobSubmitted:


    程序中调用了handleJobSubmitted方法


    上面源码中:

    1)newStage 实例化一个Stage。Stage就是一组Tasks。

    2)创建ActiveJob,是跟踪DAGScheduler中正在活动的Job

    3)判断是否在本地运行。shouldRunLocally为true调用runLocally,false则是submitStage

    接下来在本地RDD运行的Job,runLocally方法:


    这个方法启动了一个线程,里面运行了runLocallyWithinThread方法


    下面具体看下submitStage方法:


    这里是一个递归提交的过程,如果有父stage则先提交,然后将自己放置等待队列。下面网上的一张图能很好的表述这段代码:


    在你递归完所有的stage后,开始提交等待状态的stages。也就是submitWaitingStages这个方法了:


    当stage没有任何父stage依赖的时候,跟踪源码会进入submitMissingTasks。这里就开始提交任务了。提交任务的时候需要测试序列化的tasks,如果不需要则直接通过SparkListenerStageSubmitted发布tasks。最后,将会以TaskSet的方式将任务集合提交到TaskScheduler。


    TaskScheduler

    从上面的源码一路跟踪,到了TaskScheduler了。TaskScheduler有多个实现类


    这里进入到了TaskSchedulerImpl这个实现类。OK,看看它如何实现submitTasks这个方法的:


    1)new了一个TaskSetManager。TaskSetManager管理和跟踪TaskSet。失败的任务他会重新启动他,当然重启的次数是有限。这个有TaskSetManager的构造参数maxTaskFailures决定。

    2)添加任务调度模式。Spark中提供了两种调度模式FIFO和FAIR,默认是FIFO。


    具体来看看这个调度方法。


    FIFO是先进先出,这里将作业集添加到调度队列中去了。

    3)backend.reviveOffers()。这个backen是CoarseGrainedSchedulerBackend。这是一个调度器接口,他会等待Executors通过AKKA来连接他。


    接着进入DriverActor的事件处理方法中去receiveWithLogging,ReviveOffers消息调用了makeOffers()。源码如下:


    源码中有两个方法一个是resourceOffers。然后是launchTasks。

    resourceOffers方法会从workers中随机抽出一些来执行任务,然后通过TaskSetManager找出和Worker在一起的Task,最后打包成TaskDescription返回源码如下:



    接下来看launchTasks方法了。该方法中最重要的一句就是:

    executorActor(task.executorId) !LaunchTask(new SerializableBuffer(serializedTask))

    excutorActor是在CoarseGrainedExecutorBackend的RegisteredExecutor注册事件中,通过SparkDeploySchedulerBackend启动的AppClient。而在AppClient内部启动了一个AppActor,AppActor想Master发送注册APP信息。


    而在Master中对RegisterApolication事件是这样处理的:


    Worker 

    LaunchExecutor

    任务最终会发送到Worker中去处理,而接收处理的事件就是LauchExecutor了。处理过程如下面源码:


    源码最后又向Master发送了ExecutorStateChanged

    Master将时间转发给Driver,如果果是Executor运行结束,从相应的映射关系里面删除。

    CoarseGrainedExecutorBackend

    最后发布任务是CoarseGrainedExecutorBackend中的LaunchTask事件。源码如下:


    Executor

    最后的任务执行Executor!


    OK,现在进入Executor的run源码过程。







  • 相关阅读:
    使用postman模拟上传文件到springMVC的坑:the request was rejected because no multipart boundary was found
    win10 安装多个版本的jdk,如何切换
    String类的substring方法
    tomcat7.0配置CORS(跨域资源共享)
    win7下安装centos6.5后,开机无法进入选择双系统启动界面,只能启动centos的解决办法
    java位运算
    JDK源码--ArrayList浅析
    使用Jasperreporter生成入库出库单打印等报表操作
    centos6.5下安装zip格式的tomcat7和tomcat8,并同时运行
    Centos7配置文件共享服务器SAMBA三步曲(转)
  • 原文地址:https://www.cnblogs.com/huwf/p/4273377.html
Copyright © 2011-2022 走看看