Hadoop平台下公平调度机制的源码研究

Hadoop平台下公平调度机制的源码研究

MapReduce模型MapReduce是由Google发明的一种处理大规模数据的分布式编程框架，最初是由Google的工程师设计并实现。Hadoop是MapReduee计算模型的一种开源实现，用于大规模数据集的并行化分析处理。Hadoop中的作业调度是JobTracker指派任务(Tasks)到相应TaskTracker上执行的过程。

公平调度机制公平调度器按资源池（pool）来组织作业，并把资源公平的分到这些资源池里。默认情况下，每一个用户拥有一个独立的资源池，以使每个用户都能获得一份等同的集群资源而不管他们提交了多少作业。在每一个资源池内，会使用公平共享（fairsharing）的方法在运行作业之间共享容量（capacity）。用户也可以给予资源池相应的权重，以不按比例的方式共享集群。

最小共享量公平调度器允许赋给资源池保证（guaranteed）最小共享资源，这个用在确保特定用户、群组或生产应用程序总能获取到足够的资源时是很有用的。当一个资源池包含作业时，它至少能获取到它的最小共享资源，但是当资源池不完全需要它所拥有的保证共享资源时，额外的部分会在其它资源池间进行切分。

JobClient每一个job都会在用户端通过JobClient类将应用程序以及配置参数打包成jar文件存储在HDFS，并把路径提交到JobTracker，然后由JobTracker创建每一个Task并将它们分发到各个TaskTracker服务中去执行。

JobTrackerJobTracker是一个master服务，启动之后JobTracker会接收job，负责调度job的每个子任务task运行于TaskTracker上，并监控它们，如果发现有失败的task就重新运行它。一般情况应该把JobTracker部署在单独的机器上。

TaskTrackerTaskTracker是运行于多个节点上的slaver服务。TaskTracker主动与JobTracker通信，以心跳的形式定期向JobTracker发送消息或者请求任务，并负责直接执行每一个任务。

JobInProgressJobClient提交job后，JobTracker会创建一个JobInProgress来跟踪和调度这个job，并把它添加到job队列里。JobInProgress会根据提交的jobjar中定义的输入数据集（已分解成FileSplit）创建对应的一批TaskInProgress用于监控和调度Task。

FairSchedulerFairScheduler是hadoop中的作业公平调度器，主要是解决当TT发送心跳告诉JT当前的空闲slots时，希望JT分配给TT相应多个task，让TT去执行这些task。所以JT就需要一个调度器来对作业进行调度，选择出作业，然后将作业的task分配TT去执行。而hadoop中的task可以分为map,reduce,jobsetup,jobcleanup,taskcleanup这五种task。

任务调度流程

HeatBeatTaskTacker调用JobTracker中的heatBeat()方法请求心跳回应。JobTracker每次会对收到的heartbeat进行判断，如果心跳是请求新的task，且没有Job的stup与clearn任务。JobTracker将调用FairScheduler中的assignTasks()分配任务，调用该方法将得到分配给该资源的tasklist。

assignTasks先获取作业池中各作业待运行的runnabletasks总数，已经在运行的任务runningTasks总数，获取集群中的集群状态资源，获取集群中的总资源（slots）数。获得发送该心跳的TaskTracker结点的状态对象（TaskTrackerStatus）来获得该TaskTracker可执行的最大任务数。

assignTasksassignMultiple的值表示一次分配一个task还是分配一对task。通过一个循环体寻找可分配的任务，知道已达到分配的要求或者已无任务分配。退出代码如下：

if(mapRejected&&reduceRejected||!assignMultiple&&tasks.size()>0){break;}

assignTasks确定任务分配的taskType，当TaskTracker上两种taskType都可接受时，taskType的值取决于TaskTracker上运行的MapTask与ReduceTask的多少，哪种任务的数量比较少则分配哪种类型的任务。

Pool的排序根据FairScheduler的规则对各pool进行排序的主要指标：最小共享资源量（minShare）已运行的任务数（runningTasks)资源紧缺度（Needy）共享率（minShareRatio）权值比率（tasksToWeightRatio）

Pool的排序规则资源首先分配给现在占有资源量少于最小共享资源量的pool如果pool的最小共享量都没有达到，也就是Needy都为true，资源就分配给共享率低的pool如果pool的Needy都为false，资源就分配给权重比率低的pool如果minShareRatio、tasksToWeightRatio两者有一个相同，则依据pool的startTime。

Pool中Job排序在FairScheduler.java中的assignTasks中通过对PoolSchedulable进行排序之后，只是确定了分配资源给pool的优先顺序，并没有确定将资源分配给pool中资源的顺序，代码：assignTask(tts,currentTime,visitedForMap):是调用了PoolSchedulable中的任务分配方法，结果返回一个将获得该资源的task，如果不能分配返回null。

Pool中Job排序pool中的job排序规则可以由用户选择，目前支持两种排序规则：FIFO:优先级决定FairScheduler：jobweight的值决定

Pool中Job排序pool中的job排序规则可以由用户选择，目前支持两种排序规则：FIFO:优先级决定FairScheduler：jobweight的值决定

Job权重计算在对job进行排序时，job的权重是个非常重要的因素，job的权重越高获得的资源也将更多，如果job还未运行其权重默认为1.0，已运行的job的权重将与需求量、优先级有关。用户设定了weightAdjuster（不知道该起个什么名字），那么job的权值还与其已经提交的时间有关。设定有一定的时间，如果提交的时间已经超过了这个额度（duration），job的权重将乘以一个系数（factor）。

Job权重计算weight=Math.log1p(runnableTasks)/Math.log(2);weight*=getPriorityFactor(job.getPriority());

if(weightAdjuster!=null){//设置了权重改变策略

//Runweightthroughtheuser-suppliedweightAdjusterweight=weightAdjuster.adjustWeight(job,taskType,weight);}

结束语MapReduce计算模型已经被越来越多的企业的支持和应用。由于MapReduce集群的昂