讲说spark的资源调度和任务调度,基本的spark术语,这里不再多说,懂的人都懂了。。。

按照数字顺序阅读,逐渐深入理解:以下所有截图均为个人上传,不知道为什么总是显示别人的QQ,好尴尬,无所谓啦,开始吧~~

1 宽窄依赖与Stage划分:

上熟悉的图:



 

在 Spark 里每一个操作生成一个 RDD,RDD 之间连一条边,最后这些 RDD 和他们之间的边组成一个有向无环图,这个就是 DAG,Spark
内核会在需要计算发生的时刻绘制一张关于计算路径的有向无环图,也就是 DAG。有了DAG 图,Spark 内核下一步的任务就是根据 DAG 图将计算划分成
Stage,

上图:G 与 F 之间是宽依赖,所以把 G 和 F 分为两个 Stage,而 C 、D 到 F,E 到 F 都是窄依赖,所以 CDEF 最终划分为一个
Stage2,A 与 B 之间是宽依赖,B 与 G 之间是窄依赖,所以最终,A 被划分为一个 Stage1,因为 BG 的 stage 依赖于 stage1
和 stage2,所以最终把整个DAG 划分为一个 stage3,所以说,宽窄依赖的作用就是切割 job,划分 stage。
Stage:由一组可以并行计算的 task 组成。
Stage 的并行度:就是其中的 task 的数量。
与互联网业界的概念有些差异:在互联网的概念中,并行度是指可同时开辟的线程数,并发数是指每个线程中可处理的最大数据量,比如: 4
个线程,每个线程可处理的数据为 100 万条,那么并行度就是 4,并发量是 100 万;而对于 stage 而言,即使其中的
task是分批进行执行的,也都算在并行度中,比如,stage 中有 100 个 task,而这 100 个 task 分4 批次才能执行完,那么该 stage
的并行度也为 100。Stage 的并行度是由最后一个 RDD 的分区决定的。

2 资源调度 (有人喜欢那standalone client方式举例,那我这把cluster 模式和client模式都说下得了~)

Cluster模式:

源码位置如图master.scala:



a)Worker 启动后向 Master 注册



b) client 向 Master 发送一条消息,为当前的 Application 启动一个 Driver 进程



启动Driver源码如下图:



c)Driver 进程向 Master 发送消息:为当前的 Application 申请一批 Executor:



EXECUTOR创建的过程如图:



上图的过程,展示了Executor进程被启动的过程。那么下面的是简图:



总结资源调度:

1) 在默认情况下(没有使用--executor --cores 这个选项)时,每一个 Worker 节点为当前的 Application 只启动一个
Executor,这个 Executor 会使用这个 Worker 管理的所有的cores。(原因:assignedCores(pos) +=
minCoresPerExecutor);
2) 默认情况下,每个 Executor 使用 1G 内存;
3) 如果想要在一个 Worker 节点启动多个 Executor,需要使--executor --cores 这个选项;
4) spreadOutApps 这个参数可以决定 Executor 的启动方式,默认轮询方式启动,这样有利于数据的本地化。

Client模式(standalone):

     实际上我们的代码会先在Driver这个进程(我们写的spark程序,打成jar包,用spark-submit来提交,local模式
当我们的程序提交到集群上时,会加载并执行我们的jar包,并找到jar包中的main函数执行一遍,执行main所启动的这个进程名就是SparkSubmit,这个进程就是我们所说的Driver
进程; cluster模式
会在集群中找一台node,会启动一个进程执行一遍我们提交的代码,这个进程就是Driver,Driver启动之后将其信息注册到Master中,存储在
ArrayBuffer中)中执行一遍,Driver这个进程中有SparkContext这个对象。代码从main函数开始执行,new
SparkConf()设置我们运行Spark时的环境参数(还可以通过spark-submit -- 加上参数来设置),new
SparkContext()(在源码的700多行左右),创建DAGScheduler和TaskScheduler,TaskScheduler另
启动一个线程将Application注册到Master中【  是放到Master中的ArrayBuffer
的数据结构中,当ArrayBuffer中有信息之后,Master会调用自己的schedule()
方法,schedule会为当前的Application申请资源,此时master会找一些空闲的Worker,并在Worker上启动Executor
进程,Executor启动完成之后会反向注册给TaskScheduler(在driver)】 。

 

3 任务调度

Cluster模式:

源码位置:core/src/main/scala/rdd/RDD.scala。



上图是任务调度源码,不在文字叙述,可以参考client模式的部分叙述。

注意:2点:

【a:提交时的容错能力】


TaskScheduler提交task如果发生了失败,默认会重试三次,如果依然失败,那么则认为这个task就失败了,这时会进行stage重试,DAGScheduler会重新发送TaskSet给TaskScheduler,默认会重试四次,如果四次后依然失败,则认为job失败。因此一个task默认情况下重试3*4=12次

如果task失败是由于shuffle file not find造成的,那么TaskScheduler是不负责重试的,直接进行stage重试。

【b:重试机制的问题】

如果:Task的逻辑是将处理的数据结果放入到数据库中,

如果一个Task提交到百分之七十五,然后失败了,这时候会重试,那么有执行了一次task,这时候就会有脏数据的产生。

以上问题如何去解决?

【1.关闭重试机制。
    2.在数据库中设置主键,这时候如果重复提交,那么会失败,也就避免了脏数据的产生。



 

Client 模式(接2 资源调度中的Client模式):

任务调度: (此时代码的第二行(new SparkContext(conf))已经执行完毕)     

a).
Action类型的算子触发job的执行。源码中调用了SparkContext的runJob()方法,跟进源码发现底层调用的是DAGScheduler的runJob()方法。

   DAGScheduler会将我们的job按照宽窄依赖
划分为一个个stage,每个stage中有一组并行计算的task,每一个task都可以看做是一个”pipeline”,,这个管道里面数据是一条一条被计算的,每经过一个RDD会经过一次处理,RDD是一个抽象的概念里面存储的是一些计算的逻辑,每一条数据计算完成之后会在shuffle
write过程中将数据落地写入到我们的磁盘中。

 

b).stage划分完之后会以TaskSet的形式提交给我们的TaskScheduler。

    源码中TaskScheduler.submit.tasks(new
TaskSet())只是一个调用方法的过程而已。我们口述说是发送到TaskScheduler。TaskScheduler接收到TaskSet之后会进行遍历,每遍历一条调用launchTask()方法,launchTask()根据数据本地化的算法发送task到指定的Executor中执行。
task在发送到Executor之前首先进行序列化
,Executor中有ThreadPool,ThreadPool中有很多线程,在这里面来具体执行我们的task。

 

c).TaskScheduler和Executor之间有通信(Executor有一个邮箱(消息循环体CoresExecutorGraintedBackend)
),Executor接收到task

    Executor接收到task后首先将task反序列化,反序列化后将这个task变为taskRunner(new
taskRunner),并不是TaskScheduler直接向Executor发送了一个线程,这个线程是在Executor中变成的。然后这个线程就可以在Executor中的ThreadPool中执行了。

 

d).Executor接收到的task分为maptask 和 reducetask

    map task 和 reduce task,
比如这里有三个stage,先从stage1到stage2再到stage3,针对于stage2来说,stage1中的task就是map task
,stage2中的task就是reduce task,针对stage3来说...map task 是一个管道,管道的计算结果会在shuffle
write阶段数据落地,数据落地会根据我们的分区策略写入到不同的磁盘小文件中,注意相同的key一定写入到相同的磁盘小文件中),map端执行完成之后,会向
Driver中的DAGScheduler对象里面的MapOutputTracker发送了一个map task的执行状态
(成功还是失败还有每一个小文件的地址)。然后reduce task开始执行,reduce端的输入数据就是map端的输出数据。那么如何拿到map端的输出数据呢?
reduce task会先向Driver中MapOutPutTracker请求这一批磁盘小文件的地址,拿到地址后,由reduce
task所在的Executor里面的BlockManager向Map task 所在的Executor先建立连接,连接是由ConnectionManager
负责的,然后由BlockTransformService去拉取数据,拉取到的数据作为reduce task的输入数据(如果使用到了广播变量,reduce
task 或者map task 它会先向它所在的Executor中的BlockManager要广播变量,没有的话,本地的BlockManager会去连接
Driver中的BlockManagerMaster,连接完成之后由BlockTransformService
将广播变量拉取过来)Executor中有了广播变量了,task就可以正常执行了。

 

只说实用,不举理论!