网站关键词搜索优化是怎么做的,品牌咨询,广告推广的软件,软文案例目录 1、DAGScheduler与TaskScheduler
2、TaskScheduler与SchedulerBackend
3、任务调度过程总结 1、DAGScheduler与TaskScheduler
DAGScheduler面向我们整个Job划分出了Stage#xff0c;划分了Stage是从后往前划分的#xff0c;执行的时候是从前往后#xff0c;每个Stag…目录 1、DAGScheduler与TaskScheduler
2、TaskScheduler与SchedulerBackend
3、任务调度过程总结 1、DAGScheduler与TaskScheduler
DAGScheduler面向我们整个Job划分出了Stage划分了Stage是从后往前划分的执行的时候是从前往后每个Stage内部有一系列任务Stage里面的任务是并行计算的这些并行计算的任务的逻辑是完全相同的只不过是处理的数据不同而已。DAGScheduler会以TaskSet的方式以一个DAG构造的Stage中所有的任务提交给底层调度器TaskSchedulerTaskScheduler是一个接口做接口的好处就是跟具体的任务调度解耦合这样Spark就可以运行在不同的资源调度模式上Standaloneyarnmesos等这符合面向对象中依赖抽象而不依赖具体的原则带来了底层资源调度器的可插拔性导致Spark可以运行在众多的资源调度器模式上。
DAGScheduler是高层调度器TaskScheduler是底层调度器。高层调度器本身是属于Spark自己内核的实现通过划分不同的Stage这个是优化的核心。底层调度器做了一个接口面向接口了解这个接口可以自己定义底层的具体调度器。
2、TaskScheduler与SchedulerBackend
DAGScheduler完成了面向Stage的划分之后会按照顺序逐个将我们的Stage通过TaskSchedulerImpl的submitTaskScheduler提交给底层调度器调度过程如下
1DAG构建完毕提交stage给taskSchedulertaskScheduler执行submitTasks开始调起具体的Task他主要的作用是将TaskSet加到我们的TaskSetManager中这里new了一个TaskSet。 2因为TaskScheduler是个接口进入到他的具体实现也就是TaskSchedulerImpl。在这里任务集合中的任务可能是ShuffleMapTask也可能是ResultTask。我们把任务集合赋值给变量tasks然后在同步代码块中创建了一个createTaskSetManager在其内部路由了TaskSetManager。构建TaskSetManager的参数包括thisTaskSchedulerImpl本身taskSet还有最大失败重试次数是构建TaskSchedulerImpl的时候传进来的如果没设置的话默认最大为4 3在创建了TaskSetManager后比较关键的一点就是调用了SchedulableBuilder.addTaskSetmanagerSchedulableBuilder会确定TaskSetManager的调度顺序然后按照TaskSetManager的locality aware来确定每个task具体运行在哪个ExecutorBackend中 schedulableBuilder他是SchedulableBuilder类型的是整个应用程序级别调度器我们是在TaskSchedulerImpl创建的时候就对他实例化。他本身是我们应用程序级别的调度器他自己支持2种调度模式有FIFO先进先出和FAIR公平调度。我们的调度策略可以通过spark-env.sh中的spark.scheduler.mode进行具体的配置默认是FIFO。 4在SchedulableBuilder.addTaskSetmanager之后有一个关键的步骤backend.reviveOffers()他的作用就是给DriverEndPoint发ReviveOffers消息。 运行时调用的是CoarseGrainedSchedulerbackend的reviveOffers方法。ReviveOffers是个case Object啥也没有因为不需要封装数据相当于一个触发器。因为资源调度肯定是统一进行调度算法的这里只不过相当于一个信号灯。所以只要资源变化的时候ReviveOffers就会被触发因为不是通过ReviveOffers本身去调资源的。只不过是触发调资源的过程为什么要触发资源原因很简单提交新的任务过来了 5DriverEndPoint这个驱动程序的调度器收到ReviveOffers消息后调用makeOffersmakeOffers的目的是找出可以利用的executors。 创建WorkerOfferWorkerOffer是个很简单的数据结构是个case class在这边会告诉我们具体的Executor可用的资源可用资源是是cpu的core为什么不考虑内存只考虑core因为已经分配好了基本都符合要求内存现在要计算计算的核心考虑对象是cpu launchTasks参数中的Scheduler.resourceOffers是根据已经可用的计算资源为每个task具体分配计算资源。这里的Scheduler就是TaskSchedulerImpl。输入是ExecutorBackend及其上可用的Cores输出是TaskDescription的二维数组在其中确定了每个Task具体运行在哪个ExecutorBackend。
6resourceOffers是如何确定task具体运行在哪个ExecutorBackend上的呢作为参数传进来的WorkerOffer是个一维数据里面装了所有可计算的资源返回的是个二维数据每个任务他的数据本地性。
①首先开始循环遍历可用资源o - offers然后判断如果不包含这个executor要把executor加进去刚才是最新请求了一下机器有哪些可用资源executorIdToHost这是个内存数据结构 这个是数据本地性默认情况下在一个rack实际生产环境肯定是若干个rack。
②shuffledOffers把我们所有可用计算资源打散以寻求计算的负载均衡。tasks中对每个worker new了一个ArrayBuffer里面是TaskDescription每个executor可用放几个TaskDescription就可以运行多少个任务他有多少个core就可以立即分配多少个任务。所以new一个一维数组他的长度是当前机器的cores的个数决定的。现在创建的一维数组是说当前这个ExecutorBackend可以分配多少个task也就是可以并行运行多少个task。
sortedTaskSets根据调度顺序返回的是TaskSet集体的计算集合。然后遍历所有排序后的TaskSet集合如果有可用的新的executor我们会把executorAdded交给taskSet。 如果有新的ExecutorBackend分配给我们的job此时会调用executorAdded来获得最新的完整的可用计算资源 ③回到resourceOffers方法中这个是TaskSchedulerImpl的接下来这里是两层循环追求最高最优先级本地性首先是PROCESS_LOCAL不行就NODE_LOCAL不行的话就...按优先级下去。 在resourceOfferSingleTaskSet中获得executorId根据他的id获得host然后判断一下每台机器可用的计算资源PUC_PER_TASK默认为1就是每个任务默认用一条线程。然后就就开始进行总路规则的计算
如果符合要求的话就执行所以最后算的还是要靠taskSetManager的resourceOffer他获得就是每个任务具体运行在哪里返回的就是TaskDescription ④到taskSetManager的resourceOffer方法中 因为实质来讲DAGScheduler已经告诉我们具体任务运行在哪台机器上。DAGScheduler是从数据层面考虑preferedLocation的从RDD的层面确定而TaskScheduler是从具体计算Task的角度考虑计算的本地性。他们并没有冲突而且是配合在一块的所以说TaskScheduler是更底层的调度DAGScheduler是高层的调度。DAGScheduler确定数据的本地性就是确定数据在哪台机器上根据这个数据本地性确认计算要发生在哪台机器上TaskScheduler有5个数据本地性原则肯定追求PROCESS_LOCAL考虑数据是否直接在内存中。一个是高层一个是底层高层就是preferedLocation确定数据具体在哪台机器上机器上有executor底层就是知道数据本地性的情况下任务的本地性所以底层是更具体的。TaskScheduler确定executor上具体的task运行的数据具体在内存中还是节点上还是在这台机器上这完全是2码事
7回到makeOffers方法中本地两个层面一个数据本地性一个是计算的本地性确定好后就launchTasks把任务发送给executorBackend去执行。首先将二维数组进行序列化。 任务要大于128M就把任务abort丢弃 如果任务小于128M。那么启动任务启动具体任务的是CoarseGrainedSchedulerBackend去launchTask到具体的ExecutorBackend上。TaskSetManager是跟踪任务的不是发送任务的
8CoarseGrainedExecutorBackend接收到消息之后就反序列化交给executor 3、任务调度过程总结
1DAG构建完毕提交stage给taskSchedulertaskScheduler的实例TaskSchedulerImpl执行submitTasks方法将TaskSet加入到TaskSetManager中进行管理
2SchedulerBuilder.addTaskSetManager方法SchedulerBuilder确定在哪个ExecutorBackend中按照TaskSetManager的locality aware来确定每个task具体运行在哪个ExecutorBackend中SchedulableBuilder是整个应用程序级别调度器有FIFO和FAIR两种调度模式
3CoarseGrainedSchedulerBackend.reviveOffers给DriverEndPoint发送ReviveOffersReviveOffers本身是一个空的case Object对象只是起到触发底层资源调度的作用在有Task提交或者计算资源变动的时候会发生ReviveOffers这个消息作为触发器
4DriverEndPoint收到ReviveOffers消息然后路由到makeOffers具体的方法中在makeOffers方法中首先准备好所有可以用于计算的workOffers代表了所有可用ExecutorBackend中可以使用的Cores等信息
5调用TaskSchedulerImpl的resourceOffers作为参数执行launchTasks发送出去。TaskSchedulerImpl为每一个Task具体分配计算资源输入是ExecutorBackend及其上可用的Cores输出TaskDescription二维数组在其中为每个Task具体运行在哪个ExecutorBackend。 resourceOffers是如何确定Task运行在哪个ExecutorBackend上的算法实现如下
1通过Random.shuffle方法重新洗牌所有的计算资源用于负载均衡
2根据每个ExecutorBackend的Cores个数申明声明类型为TaskDescription的ArrayBuffer数组
3如果新的ExecutorBackend分配给我们的Job此时会回调用executorAdded来获得最新的完整的可用计算资源
4求的最高级别的优先级本地性
5通过TaskSetManager的resourceOffers最终确定每个Task具体运行在哪个ExecutorBackend的具体的本地性
6通过launchTasks把任务发送给ExecutorBackend执行。 注意
1默认Task的最大重试次数是4
2spark目前支持FIFO的FAIR两种调度器
3TaskScheduler中要负责为Task分配计算资源此时程序已经具备集群中你的计算资源了根据计算本地性原则确定Task具体运行在哪个ExecutorBackend中每个Task默认是采用一条线程计算的
4TaskDescription中已经确定好Task运行在哪个ExecutorBackend
5数据本地优先级由高到底为PROCESS_LOCALNODE_LOCALNO_PREFRACK_LOCALANY
6DAGScheduler是从数据层面考虑数据本地性而TaskScheduler是从具体计算Task角度考虑计算本地性的。也就是DAGScheduler通过preferedLocation确定数据的本地性就是确定数据在哪台机器上根据这个数据本地性确认计算要发生在哪台机器上TaskScheduler就是知道数据本地性的情况下任务的本地性所以底层是更具体的
7Task进行广播的时候如果任务大于128M的话则Task会被直接丢弃如果小于128M的话会通过CoarseGrainedSchedulerBackend去launchTask到具体的ExecutorBackend上。
