Spark资源调度和任务调度

发布于:2024-11-03 ⋅ 阅读:(12) ⋅ 点赞:(0)

Spark client模式下任务成功运行时会在node1、node2两个节点,随机一个节点上启动两个进程 :

1) ApplicationMaster :  所有任务在yarn上运行都会启动的进程

2) Executor :  执行器

在master节点启动一个进程 :

1) SparkSubmit   -- client模式下就是一个Driver

任务结束后这几个进程会立即消失

这是Spark向yarn申请到的资源。

1、 Spark 应用程序由两部分组成 :

    (1) Driver     -- 负责任务调度

         Spark应用程序的主程序,负责将task发送到Executor中执行(写好的代码最终会变成一堆task, maptask、reducetask)

(2) Executor   -- 负责执行task

         负责执行task,并将执行状态汇报给Driver(task是成功还是失败)

 需要自己来指定  默认1个 1核 1G   (1核  -- 同一时刻只能执行一个task)

 修改为 2个 2核 1G :    2个Executor分散在node1、node2两个节点中 :  

 spark-submit --class 类名 --master yarn-client --num-executors 2 --executor-memory 1G --executor-cores 2 架包

     yarn一共只有16G 16核,所以不能将Executor设置太多个

Driver、 Executor 本质上都是一个JVM(java进程)

当在client模式下提交一个任务时,会在本地(提交任务这台服务器,并非本电脑。 例如在master上提交的,master就是本地)启动一个Driver。

    详细日志都是Driver打印出来的。

当在IDEA本地"local"模式下(单机)运行代码,Driver和Executor合并了,是同一个东西。

只有集群模式才分开。

2、 yarn-c

    (1) client模式   

1、在本地启动Driver

2、Driver向yarn的ResourceManager申请资源,申请启动ApplicationMaster(AM是所有在yarn上运行都会启动的进程)

3、RM分配一个节点NodeManager启动AM

4、AM向RM申请启动Executor

5、RM分配一批节点启动Executor

6、Executor反向注册给Driver(因为是Driver将task发送到Executor)

7、当资源申请到之后,Driver将task发送到Executor执行

yarn-client模式在本地启动Driver,Driver会和集群中的Executor进行网络通信,如果我们在本地提交多个spark任务,

会导致本地网卡流量剧增。

-- 一般用于上线的测试,可以在本地看到详细的执行日志。

(2) cluster模式

        1、Driver不在本地启动,而是随机启动在集群中的一个NodeManager中

2、向RM申请启动AM

3、RM分配一个节点NodeManager启动AM

4、AM向RM申请启动Executor

5、RM分配一批节点启动Executor

6、Executor反向注册给节点中的ApplicationMaster(cluster模式下,AM相当于Driver)

yarn-cluster模式Driver随机启动在集群中,不会导致某一台服务器网卡流量剧增。

-- 一般上线使用,在本地看不到详细的执行日志。

两个模式的区别 :

        client模式  Driver在本地启动

    cluster模式 Driver不在本地启动

3、 名词解释

(1) Application :  基于Spark的应用程序,包含了Driver程序和集群上的Executor

(2) Executor :  是在一个WorkerNode上为某应用启动的一个进程,该进程负责运行任务,并且负责将数据存在内存或者磁盘上。

                每个应用都有各自独立的Executors

(3) Task :  是一个线程对象, 是被送到某个Executor上的执行单元

(4) DriverProgram :  运行main函数并且新建SparkContext的程序

(5) ClusterManager :  在集群上获取资源的外部服务 ( 例如 standalone,Mesos,Yarn )

(6) WorkerNode :  集群中任何可以运行应用用代码的节点

(7) Job :  包含很多任务的并行计算的task(一个job中有很多个task),可以看做和Spark的Action (操作算子) 对应,

           每个Action都会触发一个Job任务

(8) Stage :  一个Stage是一组可以并行计算的task    。 一个Job会被拆分成很多组任务,每组任务被称为Stage。

             就像MapReduce分 map任务和reduce任务一样, Spark中的Stage可以理解为是一个map端或一个reduce端(shuffle前为map端,shuffle后为reduce端)

关联 :

     一个IDEA代码运行的程序是一个Application,一个Application里面包含了多个Job,

 一个Job是由一个Action算子处触发的,同时一个Job里面包含很多的Stage,

 一个Stage是一组可以并行计算的task(Spark中一个Stage可以理解为一个map端或一个reduce端)

 yarn-client模式和yarn-cluster模式区别

划分task

4、 资源调度 / 任务调度

    所有的大数据计算引擎都是由两部分组成 :

    (1) 资源调度

    (2) 任务调度

 任何一个大数据引擎(MapReduce、Spark、Flink...)都会先做 资源调度   --   申请资源

    (1) Spark     --  粗粒度资源调度 (即在任务调度之前将所有需要的资源全部向yarn申请下来)

  资源调度和任务调度是 分开完成的,先资源调度再任务调度。

优点 :  task在执行时不需要单独申请资源。

task执行变快了,整个Job就变快了。  所以Spark比MR快。

缺点 :  只有当最后一个task执行完才会释放资源,会导致资源浪费。 因为Spark资源占用的时候,其它任务会处于等待状态。

yarn的资源 :  总的内存 和 总的核数

(2) MapReduce --  细粒度资源调度 (即mr每一个task在执行之前单独向yarn申请资源)

                  资源调度和任务调度是 同时进行的。    

优点 :  不会占用额外的资源,不会导致资源浪费。

缺点 :  task启动会变慢。 因为task在启动之前需要先申请资源。

Spark 比 MapReduce 快的原因 :

    Spark是粗粒度资源调度

    MapReduce是细粒度资源调度

5、DAG有向无环图  

       用于描述RDD的转换关系

   RDD的转换图在Spark中被称为DAG有向无环图。

6、任务调度流程 :   

   在Driver上进行的 :

    (1) 先构建DAG有向无环图     -- 根据写的算子代码构建

(2) RDD Object会将DAG有向无环图发送给DAGScheduler     -- DAGScheduler会根据宽窄依赖切分Stage

(3) DAGScheduler会将切分好的每一个Stage,按照执行顺序(先发送map端的)以TaskSet的形式发送给TaskScheduler  

             -- Set 集合,唯一无序。保证了同一个task不重复执行多次(去重)。   

   在Executor上进行的 :

(4) TaskScheduler将Task发送给Executor    -- 通过网络发送

(5) task在Executor的线程池中执行   -- Task是个线程对象,会先在线程池中启动起来

(6) Executor会将执行结果返回给TaskScheduler,TS也会返回给DAGScheduler。

    当DS中map端的Stage都执行完,再发送reduce端的Stage,再重新完成一遍(3)(4)(5)(6)步骤。

当所有的Stage都执行完毕,一个job就执行完成了。

Driver与Executor是通过网络来通信的

Task是一个线程对象,里面是我们写的算子的代码。  

即RDD 算子内的代码 都会被封装成一个task去Executor端执行。所以Task可能会失败,失败会有重试机制,不会立马失败。

    重试机制 :

(1) 如果Task失败,TaskScheduler会重试3次

(2) 如果还是失败,DAGScheduler会对Task所在的Stage重试4次

如果还失败,整个job就失败了

(3) 如果是因为shuffle过程中文件找不到导致的异常,那么TaskScheduler就不负责重试后面的Task

而是由DAGScheduler重新上一个Stage

推测执行 :

    如果有的Task执行很慢,TaskScheduler会发送一个一模一样的Task到其他节点中执行,让多个Task竞争,

谁先执行完成以谁的结果为准。

任务调度就是把这些Task启动起来,通过代码逻辑处理HDFS中的数据。

Spark先执行资源调度,再执行任务调度。

 任务调度流程图

7、 资源调度和任务调度   -- 以yarn-client模式为例:

   一、 资源调度

    (1) 在本地启动Driver,向ResourceManager申请资源,申请启动ApplicationMaster(AM是所有在yarn上运行都会启动的进程)

(2) RM分配一个节点NodeManager启动AM

(3) AM向RM申请启动Executor

(4) RM分配一批节点启动Executor

(5) Executor反向注册给Driver端的TaskScheduler

当遇到一个Action算子的时候开始任务调度

   二、 任务调度

    (6) 先构建DAG有向无环图  

(7) DAGScheduler根据宽窄依赖划分Stage,每一个Stage是一组可以并行计算的Task

(8) DAGScheduler按照执行顺序,将Stage以TaskSet的形式发送给TaskScheduler

(9) TaskScheduler将task发送到Executor中执行

(10) task在Executor的线程池中执行,执行完成会反馈给TaskScheduler     -- 一个job完成

重试机制 :

(1) 如果Task失败,TaskScheduler会重试3次

(2) 如果还是失败,DAGScheduler会对Task所在的Stage重试4次

如果还失败,整个job就失败了

(3) 如果是因为shuffle过程中文件找不到导致的异常,那么TaskScheduler就不负责重试后面的Task

而是由DAGScheduler重新上一个Stage

推测执行 :

    如果有的Task执行很慢,TaskScheduler会发送一个一模一样的Task到其他节点中执行,让多个Task竞争,

谁先执行完成以谁的结果为准。

资源调度和任务调度