Hadoop调度算法相关研究

现在，Hadoop自带的调度策略规定是先进先出(FIFO)的，很多系统也是直接用它。虽然FIFO策略简单稳定，但随着用户和服务的日益增多，特别是服务等级的区分日益明显，高资费的用户希望拥有更优先的服务，因此FIFO没有办法适应越来越多的Hadoop商业应用需求。相关的开发种也有人考虑队列容量分配和公平队列算法，但算法实现都不够实用，也没有认真分析Hadoop中服务优先区分的具体要求。因此,本文将重新考虑其调度策略,在考虑服务等级区分和尽量保证公平的想法基础上，提出新的调度算法,并编码实现。

由于Hadoop的MapReduce调度是由tasktracker主动向jobtracker请求的，其原理类似于普通的非抢占式操作系统调度，即任务一旦分配，就不可中断。根据调研，已有典型调度算法如下：

1).先进先出算法(FIFO:Fisrt In First Out):该算法按照进程进入就绪队列的先后顺序来选择。即每当进入进程调度，总是把就绪队列的队首进程投入运行。Hadoop自带的调度算法就是FIFO。

2).时间片轮转算法(RR:Round-Robin):主要是分时系统中使用的一种调度算法。轮转发的基本思想是：将cpu划分成一个个时间片，就绪队列中的就绪进程轮流运行一个时间片。在Hadoop，可以将每个task请求看作一个时间片，用轮转的方法来调度可以使得所有job的task轮流被调度。

3).最高优先级算法(HPF:Height Priority First)：该算法调度进程时，每次将处理即分配给具有最高优先级的就绪进程。进行优先数的设置可以时静态的，也可以是动态的。静态优先数是在进程创建是根据进程初始特性或用户要求而确定的，在进程运行期间不能再改变。动态优先数则是指再进程创建时先确定一个初始优先数，以后可以在进程运行中随着进程的特性改变。

4).加权轮循算法(WRR:Weighted Round Robin)：在网络路由领域，不少研究考虑了加权的轮循算法，本文将该算法思想引入Hadoop的任务调度中，结合Hadoop任务处理的特性，提出了适合于该平台的特有的基于优先级的加权轮转算法。

下面重点介绍一下基于优先级的加权轮转算法

基于优先级的加权轮转算法

考虑到TaskTracker主动请求task的模式和hadoop任务调度体系非抢占式的特点；为了使调度的任务避免长期的等待，同时各个任务调度优先级又能够根据实际情况调整，我们将基于加权轮转调度算法的基本思想，综合考虑hadoop网络调度需求的实际情况，提出适合Hadoop任务调度的基于优先级的加权轮转调度算法(Priority Based Weighted Round Robin,PBWRR)。

算法基本思想为：将各个待运行的job放入一个队列，在不加权的情况下，轮流将任务的task交给tasktracker去执行。加权的情况下，权重较大的job可以在一次轮流中执行多个tasks。

算法的基本步骤是：

1).在tasktracker资源可用的情况下，tasktracker主动向jobtracker提交任务分配请求。

2).Jobtraker接受到tasktraker的任务分配请求后，考虑调度队列中当前job的一个task交给请求的tasktracker去执行，更新该job的剩余task信息，同时将该job的thisRoundTask值减1，如果减1后thisRoundTask结果小于1，将指针移动到下一个job；否则指针不动，等待下一个tasktracker请求的到来。

3).当指针到达队尾时，更新整个队列的相关信息，如果有job完成工作或者新的job加入，则重新计算每个job的thisRoundTask值,并将指针移动到队列的开始。

jobQueue的调度队列中每个元素jobInfo的数据结构为：

Class jobInfo
{
Int JobId;
Int jobSize;
Int taskNum;
Int meanTaskSize;
Int priority;
Int weight;
Int ThisRountTask;
}

整个调度中涉及两种队列，jobQueue和taskQueue[].其中jobQueue的元素为jobInfo,每个taskQueue中的所有元素对应一个job的所有map或reduce的task。在本文编程实现中,所有queue均使用数组实现。

分配任务的方法assignTasks具体算法下：

synchronized List<Task>assignTasks(TaskTrackerStatus tracker)
{
1)．首先考察整个jobQueue中是否有等待处理的map task或reduce task任务，如果没有返回null。
2)．考察当前jobQueue中指针指向的jobInfo，根据该jobInfo的JobId找到对应的taskQueue[i]。
3)．考察其为maptask Queue还是Reducetask Queue，如果是Reducetask Queue，需要考虑其对应的maptask是否已完成，如果其对应的maptask没有完成，需要将jobQueue的指针后移，重新考虑步骤2。如果是maptask Queue或者是Reducetask Queue且其对应的maptask已经完成，则分配该Queue的队首task给请求任务的tasktracker。
4)．从分配task的taskQueue中删除该task并更新该taskQueue的统计信息，更新jobQueue中对应的jobInfo信息（包括thisRoundTask-1,doneTask+1等）。如果更新后的jobInfo的thisRoundTask值小于1，则将jobQueue指针后移，否则指针保持不动。
5)．如果指针已经移动到队尾，则根据jobQueue更新后的jobInfo信息，如果有job完成任务或者有新的job加入，则重新计算所有的jobweight，即每个job的thisRoundTask值，再将指针移动至队首。
6).返回步骤3所分配的task。
}

其中步骤5中的权重及其thisRoundTask计算方法如下：

updateJobQueue(jobQueue queue1)
{
weight＝calculateWeight（priority）;
thisRoundTask=Min（floor(weight)，taskNum）;
}

其中priority的取值在运行时配置，由于云计算提供的业务种类繁多，收费和服务质量要求也各不相同，因此priority的取值也比较丰富。但在新增业务及其priority时，必须参考以前所有业务的priority取值，从而保持整个系统的相对公平性。当然，这应该时在业务规划时考虑的问题，而不是本文思考的重点。在本文的实验中，我们就简单的将priority取值为1、2、3等整数值。thisRoundTask为weight下取整和该job剩余的taskNum中取较小的值。

权重计算方法推导过程:

每个job[i]的meanTaskSize[i]=jobSize[i]taskNum[i];如果job[i]的权重为weight[i],则整个系统一个调度周期处理的数据量为：

同时，如果每个job[i]有优先级priority[i],为了保证其优先级的有效性，我们应该使得在这个周期中该job的数据处理量占整个系统数据处理量的百分比基本等于该job优先级值占整个系统优先级值之和的百分比，即：

其中priority[i]和meanTaskSize[i]都是已知的，我们唯一需要考虑的就是整个系统一个周期处理的数据量S。

考虑系统的处理能力，由于系统应用的绝大多数task的size都应该为blockSize，因此我们考虑整个系统所支持的MapTask能力（因为一般情况下ReduceTask需求小于mapTask）taskAbility，因此S可以定义为公式5.4

其中num表示一个周期内平均每个job执行的task数量，jobNum为待处理

的job个数，在本文实验中，我们对num取值2。

假设整个系统有m个pc组成，第i个pc可以支持t[i]个tasktracker，每个tasktracker的单位处理能力为pt[i],整个系统的单位处理能力就为：

其中p[i]为这台pc分配给Hadoop的整个处理能力，因为在单核计算机中，CPU是进程时分复用的，因此每个tasktracker的处理能力和每个单位tasktracker的处理能力的乘积就等于pc分配给Hadoop的整个处理能力p[i]。我们通过P值和S值可以估计系统一个周期运行的大致时间，但由于调度和实际任务的size问题，实际运行时间和估计时间由一定差距。