Spark Core_资源调度与任务调度详述

时间:2023-12-27 12:22:55

转载请标明出处http://www.cnblogs.com/haozhengfei/p/0593214ae0a5395d1411395169eaabfa.html


Spark Core_资源调度与任务调度详述

资源调度与任务调度(standalone client 流程描述)
    集群启动后,Worker会向Master汇报资源情况(实际上将Worker的资源写入到Master的HashSet数据机构中)
    一个 Worker 默认给一个 Application 启动 1 个 Executor,可以设置 --executor-cores num 来启动多个。开机启动时最好设置 spreadOut, 可以在集群中分散启动 executor。 
资源调度:
    实际上我们的代码会先在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 -- 加上参数来设置),newSparkContext()(在源码的700多行左右),创建DAGSchedulerTaskScheduler,TaskScheduler另启动一个线程将Application注册到Master中(是放到Master中的ArrayBuffer的数据结构中,当ArrayBuffer中有信息之后,Master会调用自己的schedule()方法,schedule会为当前的Application申请资源,此时master会找一些空闲的Worker,并在Worker上启动Executor进程,Executor启动完成之后会反向注册TaskScheduler
现在Application有了,有了资源后,开始进行任务的调度
任务调度: (此时代码的第二行(new SparkContext(conf))已经执行完毕)     
1.Action类型的算子触发job的执行。源码中调用了SparkContext的runJob()方法,跟进源码发现底层调用的是DAGScheduler的runJob()方法。
   DAGScheduler会将我们的job按照宽窄依赖划分为一个个stage,每个stage中有一组并行计算的task,每一个task都可以看做是一个”pipeline”,,这个管道里面数据是一条一条被计算的,每经过一个RDD会经过一次处理,RDD是一个抽象的概念里面存储的是一些计算的逻辑,每一条数据计算完成之后会在shuffle write过程中将数据落地写入到我们的磁盘中。
2.stage划分完之后会以TaskSet的形式提交给我们的TaskScheduler。
    源码中TaskScheduler.submit.tasks(new TaskSet())只是一个调用方法的过程而已。我们口述说是发送到TaskScheduler。TaskScheduler接收到TaskSet之后会进行遍历,每遍历一条调用launchTask()方法,launchTask()根据数据本地化的算法发送task到指定的Executor中执行。task在发送到Executor之前首先进行序列化,Executor中有ThreadPool,ThreadPool中有很多线程,在这里面来具体执行我们的task。
3.TaskScheduler和Executor之间有通信(Executor有一个邮箱(消息循环体CoresExecutorGraintedBackend)),Executor接收到task
    Executor接收到task后首先将task反序列化,反序列化后将这个task变为taskRunner(new taskRunner),并不是TaskScheduler直接向Executor发送了一个线程,这个线程是在Executor中变成的。然后这个线程就可以在Executor中的ThreadPool中执行了。
4.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里面的BlockManagerMap task 所在的Executor先建立连接,连接是由ConnectionManager负责的,然后由BlockTransformService去拉取数据,拉取到的数据作为reduce task的输入数据(如果使用到了广播变量,reduce task 或者map task 它会先向它所在的Executor中的BlockManager要广播变量,没有的话,本地的BlockManager会去连接Driver中的BlockManagerMaster,连接完成之后由BlockTransformService将广播变量拉取过来)Executor中有了广播变量了,task就可以正常执行了。

【细节一:提交时的容错能力】

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

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

【细节二:重试机制的问题】

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

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

如何去解决?

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