基于奖惩共存收益模式的大数据作业调度器

胡 静

(山西农业大学 软件学院，山西 晋中 030801)

0 引 言

目前大数据计算平台中已经存在的调度器如默认的FIFO调度器、公平调度器和计算能力调度器[1]等其它算法[2,3]调度作业时只考虑到了平台资源利用率但没有考虑作业截止时间和服务商收益问题。针对作业截止时间要求，部分研究学者[4-9]以作业截止期限为约束提出一种相应的作业调度算法以提高作业的执行效率与平台资源利用率，但仍没考虑服务商的收益问题。文献[10-14]提出相关调度算法目的是使服务商的利润最大化。文献[15-17]均提出了新的调度方法，保证服务商的最大收益、服务的截止时间和平台的资源利用率。在服务商指定时间内完成每个作业可获得的收益模式和现有作业调度器下，服务商没有考虑到作业截止时间的准确性和作业执行的迫切需求，并且也没有考虑作业调度结果对平台资源利用率的影响。

基于上述问题，本文提出了一种服务商收益模式—奖惩共存收益模式(RP Model)。该收益模式考虑用户对服务商的奖励政策，即当服务商在作业截止时间之前的一定时间范围内完成作业时对服务商给予相应的奖励值。此收益模式针对的是不确定作业截止时间是否准确的用户，并且该类用户对于作业执行完成有迫切需求。服务商为了满足用户较短的作业完成时间需求，尽量缩短每个作业的完成时间。当作业实际完成时间比用户提供的截止时间短时，服务商将该作业的实际完成时间作为较准确的截止时间反馈给用户，使得用户以后提交相同的作业时就可以提供较准确的截止时间，并且推进了用户处理作业群的整体行为。奖惩共存收益模式的提出实现了服务商与用户的利益共赢。

本文以奖惩共存收益模式为目标，提出了一种高效的MapReduce作业调度器——最大收益完成时间资源利用率调度器(MPCRS)，以尽量缩短每个作业的完成时间，提高平台资源利用率，使服务商获得最大收益。

1 MPCRS调度器设计

本文是在Hadoop2.x的Yarn资源管理系统的基础上提出的MPCRS。Yarn作为统一资源管理系统，核心组件为全局资源管理器(resource manager，RM)，负责整个平台的资源管理和分配。它主要由两个组件构成：作业调度器(Scheduler)和应用程序管理器ApplicationsManager(ASM)。调度器能够根据容量、队列等限制条件，将平台中的资源按照作业资源需求分配给各个将要运行的作业。由于调度器是一个可插拔的组件，故本文将以满足服务商最大收益、平台最大资源利用以及作业最短完成时间为目标设计一种MPCRS作业调度器。图1展示了在Yarn平台中MPCRS的作用以及构成。

图1 MPCRS的作用与构成

如图1所示，该平台允许多用户提交多个带有SLA约束的作业。RM中主要包含MPCRS和ASM两种组件，其中MPCRS是由RP Model、基于任务执行时间的确定轮数算法(TRN)以及基于最大轮数的作业调度算法(MRNS)组成。RP Model将每个作业在不同时间段内可获得的相应收益值作为TRN算法与MRNS算法的输入值。为了尽量缩短每个作业的完成时间，TRN算法以RP Model下可获得的收益值为标准，根据各个作业的Map和Reduce任务执行时间，确定出作业在不同奖惩阶段的Map和Reduce最大轮数组合以及最大标准时间。MRNS算法得到最大轮数组合方案和最大标准时间值后，结合平台中现有资源数量，考虑平台资源利用率，制定出对于当前所有作业的调度策略(TS)。ASM将会根据TS策略启动相应作业的(application master，AM)，AM向RM申请所需的资源，RM为其任务分配在各个(node manager，NM)上的Container资源，使得该作业得以开始运行。AM将作业分为Map和Reduce任务，由于MPCRS产生的TS策略是基于任务级别的，故为作业分配资源即为Map或Reduce任务分配资源。

2 MPCRS调度器实现

本文提出的MPCRS调度器是生成相应的作业调度策略，而具体任务分配方法仍然遵循MapReduce的动态分配原则，因此不会对平台中负载均衡等其它性能特性造成影响。在满足平台资源利用率最大的条件下，虽尽量缩短每个作业的完成时间，但不可避免会造成个别作业超出截止期限，无法按时完成。出现这种情况时需根据服务商收益最大化的目标对作业的收益与赔付代价进行衡量评估，经过整体收益权衡后选择放弃一些收益较少或赔付较少的作业。

2.1 奖惩共存收益模式-RP Model

在计算平台中认为每个节点的处理能力都大致相同，由于平台中的资源是统一进行管理和分配，故可动态分配的Container资源用R表示。用户提交的作业j已知如下信息：j的截止时间j_deadline；在j_deadline之前j被完成，服务商可获得的收益j_profit；如果j的实际完成时间j_completion的a倍比j_deadline短，用户按照SLA中规定的奖励比率α，支付除收益以外的奖励值j_profit×α×a；j的实际完成时间j_completion超过j_deadline时，按照服务商与用户签订的SLA赔付比率β进行赔偿，即当j_deadlineb×j_deadline， 则需要赔付给用户原收益的γ倍，即j_profit×γ；j具有的Map数量j_Nm和Reduce数量j_Nr；j中每一轮Map任务的执行时间j_Tm和每一轮Reduce任务的执行时间j_Tr。为了提高作业的执行效率和平台资源利用率，在平台资源充足的条件下，要尽量使同一个作业的多个任务并行执行，而每次并行执行任务的时间称为一轮任务的执行时间。

根据以上关于作业的收益与赔付信息，结合平台中可分配的资源数量，RP Model如下

(1)

(2)

式(1)表示完成所有作业后可获得的收益和，其中每个作业的实际收益是由该作业在截止时间之前完成的收益j_profit与额外的奖励或者惩罚f(j_profit) 组成。式(2)确保了将要运行的任务每次并行请求资源时资源数不超过平台中可动态分配的资源总和。奖惩共存函数f(j_profit) 表示如下

(3)

其中，α是在作业实际完成时间比规定的截止期限短a倍时可获得的奖励比率，β是作业实际完成时间比规定的截止期限长但没有超过截止时间的b倍时作业的赔付比率，γ是作业实际完成时间超过规定截止时间b倍时所要赔付的收益倍数。在考虑到尽量使每个作业都在截止期限之前完成，故设置赔付率值要比奖励率值大，即α≤β≤1； 而作为长时间超出截止期限的惩罚设置γ≥1。 由于不能简单的根据作业实际完成时间和截止时间值就规定用户付出奖励或服务商付出赔付，而应该是在两者差值超过一定范围后用户才需要支付额外的奖励值或服务商支付额外的赔付值，故设a,b>1。

2.2 基于任务执行时间的确定轮数算法——TRN

TRN算法根据每个作业的Map和Reduce任务执行时间，以RP Model中可获得的收益值为标准，确定每个作业在不同奖惩阶段的Map和Reduce最大轮数组合以及最大标准时间。在上节中介绍了用户提交作业时已知的信息，其中包括作业j中每一轮Map任务的执行时间j_Tm和每一轮Reduce任务的执行时间j_Tr，并且j具有的Map数量j_Nm和Reduce数量j_Nr。

TRN算法的结果是要得到每个作业在不同奖惩阶段时的任务最大轮数组合方案与最大标准时间。其中有关确定任务轮数的因素具体包括

(4)

(5)

(6)

在全部动态可分配资源只分配给一个作业的情况下，RN_m是作业j的Map任务的最少执行轮数；RN_r则是Reduce任务的最少执行轮数；式(6)表示作业j的最短执行时间tleast是由任务的最少轮数与每轮任务的执行时间组成。

算法1展示了如何在各奖惩阶段获得任务最大轮数组合方案以及最大标准时间。首先，算法根据作业的已知信息计算了每个作业的Map和Reduce任务的最少执行轮数、最短执行时间tleast以及在每个奖惩阶段的最大标准时间tstandard(line(2)-line(5))； 其次，在每个最大标准时间时，比较了作业最短执行时间与最大标准时间的大小，根据奖惩共存函数f(j_profit) 的模式要求，最大标准时间的最小值一定是大于等于最短执行时间，故在每个奖惩阶段都可以得到一种轮数组合方案A，最大限度地以空间换时间方案，即 (RN_m,RN_r,tstandard-tleast)(line(8))； 最后，根据剩余时间值的大小，找到能够满足执行一轮Map和一轮Reduce任务时间的最大变量值k，i，从而得到方案集 (planjm_t∪planjr_t) ——最大限度的以时间换空间方案。由于作业完成时间超过截止时间后服务商就要对用户进行赔偿，故要尽量保证每个作业的完成时间能够达到最短，而任务的执行轮数会直接影响到作业的完成时间。同一奖惩阶段中，在轮数不超过任务数量的前提下，即 (RN_m+k)

算法1：TRN algorithm

Input：j_Tm: the processing time of the round map task

j_Tr: the processing time of the round reduce task

j_Nm: the number of map tasks for a job

j_Nr: the number of reduce tasks for a job

f(j_profit): the reward or the punishment of a job

N: the number of jobs

R: dynamically allocate the number of resources

Output：Planj_t: the maximum round numbers plans

tstandard: the maximum standard time for a job

(1)forj=0;j

(2)RN_m←the minimum round numbers of map tasks

(3)RN_r←the minimum round numbers of reduce tasks

(4)tleast←the minimum processing time of a jobj

(5)tstandard∈T←the maximum standard time of every PR stage

(6)whiletstandarddo

(7)iftstandard≥tleastthen

(8) getting the round numbers plan A (RN_m,RN_r), the remaining time (tstandard-tleast)

(9)k,i≥1andk,iareintegersandk,iarethemaximumvaluesthatmeettherequirements

(10)if(tstandard-tleast)≥j_Tmkthen

(11) getting plan B (RN_m+k,RN_r), the remaining time (tstandard-tleast-j_Tm×k)

(12)if(tstandard-tleast-j_Tm×k)≥j_Tr×ithen

(13) getting plan C (RN_m+k,RN_r+i), the remaining time (tstandard-tleast-j_Tm×k-j_Tr×i)

(14)elsethen

(15)planjm_t={A,B}

(16)endif

(17) planjm_t={A, B, C}

(18)endif

(19)if(tstandard-tleast)≥j_Tr×ithen

(20) getting plan D (RN_m,RN_r+i), the remaining time (tstandard-tleast-j_Tr×i)

(21)if(tstandard-tleast-j_Tr×i)≥j_Tr×kthen

(22) getting plan E (RN_m+k,RN_r+i), the remaining time (tstandard-tleast-j_Tr×i-j_Tm×k)

(23)elsethen

(24) planjr_t={A, D}

(25)endif

(26) planjr_t={A, D, E}

(27)endif

(28)Planj_t={A}∪planjm_t∪planjr_t

(29)endif

(30)endwhile

(31)endfor

2.3 基于最大轮数的作业调度算法—MRNS

在TRN算法确定出Map、Reduce任务最大轮数组合方案和最大标准时间后，MRNS算法结合平台中现有资源数量，考虑平台资源利用率，制定出对于当前所有作业的调度策略，以实现收益最大化的目标。MRNS算法如下所示：

算法2：MRNS algorithm

Input：Planj_t: the maximum round numbers plans

tstandard: the maximum standard time for a job

N: the number of jobs

δ: the threshold of the resource utilization

Output： TS: the optimal scheduling strategy based on tasks

F: the maximum profit

(1)forj=0;j

(2) whenj_tstandard=j_deadline/a, calculate f(j_profit)

(3)F=F+j_profit

(4)endfor

(5)F=F+max(f(j_profit))

(6)jobjof the maximum reward is added to scheduling queue

(7)whenj_tstandard=j_deadline/a,Planj_tof the jobjis selected

(8)whilePlanj_tdo

(9) recording the completion timej_completion

(10) calculating the remaining resourcesrwhen the jobjbegins to be scheduled till the end

(11)if(R-r)/R≥δthen

(12)idlingtheremainingresourcesandwaitingforthepreviousjobcompletion

(13)fori=0;i

(14)ifj_completion≥i_deadlinethen

(15) the jobiis added to the end of scheduling queue

(16)elseifi_deadline-j_completion≥i_1tstandard+j_completionthen

(17) the jobigets the reward f(i_profit)

(18)elseifi_2tstandard+j_completion≤i_deadline-j_completionthen

(19) the jobigets the reward 0

(20)elseifi_3tstandard+j_completion≤i_deadline-j_completionthen

(21) the jobigets the punishment f(i_profit)elsethen

(22) the jobigets the maximum punishment f(i_profit)

(23)endfor

(24) according to f(i_profit) for each job, sortingN-1 jobs

(25)if∀f(i_profit)≥0then

(26) the jobithat has the maximum reward is added to the scheduling queue

(27)elseif∀f(i_profit)<0then

(28) according to sorting results, find the jobimee-ting f(i_profit)>(-i_profit×γ) & the minimum f(i_profit)

(29) the jobiis added to the scheduling queueelsethen

(30) ∀f(i+n_profit)<0,f=|-i+1_profit×γ|+|-i+2_profit×γ|+…+|-i+n_profit×γ|

(31)iff(i_profit)≥fthen

(32) the jobiis added to the scheduling queueelsethen

(33) find the jobi+nmeeting f(i+n_pro-fit)=-i+n_profit×γthat is the minimum

(34) the jobi+nis added to the scheduling queue

(35)elsethen

(36) when the scheduled jobjhas the remaining resources, recording the start timetrfor the remaining resources

(37)fori=0;i

(38)iftr≤i_1tstandardthen

(39) the remaining resourcesrmeetsPlani_tthat on thei_1tstandard

(40) Comparingtr+i_1tstandardwithi_tstandard

(41) the jobiget the maximum |f(i_profit)| besides the maximum punishment

(42)ifi_1tstandard≤tr≤i_2tstandardthen

(43) the remaining resourcesrmeetsPlani_tthat on thei_2tstandard

(44) Comparingtr+i_2tstandardwithi_tstandard

(45) the jobiget the maximum |f(i_profit)| besides the maximum punishment

(46)endfor

(47) the jobiis scheduled that has the maximum |f(i_profit)|

(48)F=F+f(i_profit)

(49)endwhile

(50)F=F+f(i_profit)

算法2中首先根据所有作业在j_tstandard=j_deadline/a时的最大奖励值f(j_profit)，确定被调度的作业并且选择最大轮数组合方案集(line(1)-line(7))；其次，在前一作业的所有轮数组合方案集中，选择具有局部最大收益的轮数组合方案进行调度(line(8)-line(49))；最后将所有作业加入到调度队列后，根据每个作业相应的任务轮数组合方案，计算全局的最大收益值(line(50))。其中，在选择具有局部最大收益的轮数组合方案时，优先考虑平台的资源利用率。当资源利用率大于给定阈值时，则空闲资源，直到前一作业执行完成后，在剩余作业中选择局部收益最大的作业进行串行调度(line(11)-line(34))。在选择局部收益最大的作业时，通过每个作业的截止时间与前一作业的完成时间差所在范围，比较每个作业在该时间范围内可获得的奖励或惩罚值，选择出使得局部收益最大的作业和相应的任务轮数组合方案。当资源利用率小于给定阈值时(line(35)-line(48))，将要调度的作业与前一作业并行执行，充分利用平台资源，使得资源利用率达到最大。在选择合适的作业加入到调度队列之前，记录前一作业有空余资源时的开始时间点tr，tr与每个作业的最大标准时间tstandard进行比较，当剩余资源能够满足最大标准时间范围内的任务最大轮数要求时，则选择 |f(j_profit)| 值最大的作业加入到调度队列，对于惩罚值已经达到最大的作业则放在调度队列最后进行调度。由于每个被选中的作业都有多种最大轮数组合方案，故在每选择出一种调度组合方案后则计算相应的局部收益值，最终选择全局收益值最大的调度队列与调度策略进行调度。调度策略中包含了已选择的作业任务轮数与开始时间点，在调度前就预先指定了对于每个作业中任务的资源分配数与分配时间点。

3 实验评估

3.1 实验环境与设置

在基于Hadoop框架的计算平台中对本文提出的调度器进行验证，使用的是Hadoop 2.7.1版本。平台中的所有节点都是同构节点，其中包含1个主节点和20个从节点。每个节点的配置信息为CPU 8 cores，16 GB RAM，1 TB hard disk, Red Hat Enterprise Linux6.2 System。资源管理和调度使用Yarn资源框架，设置Container大小为1 core和2G RAM，这样每个节点上有8个Container，整个平台中共有160个Container。在实验中设置块大小为默认的64 M。

3.2 实验数据集与实验指标

为评估算法性能，使用PUMA benchmark suits中的MapReduce标准应用程序，每类程序的数量、输入数据规模、截止时间以及数据来源见表1。

表1 数据集

WordCount是CPU密集型程序，TeraSort是I/O密集型程序，Inverted-Index既是CPU密集型也是I/O密集型，3种程序均为典型的MapReduce程序。

在评估MPCRS的效率时，将MPCRS分别与FIFO、Fair、EDF调度算法在作业完成时间、服务商收益及平台资源利用率等方面进行比较。通过以下性能指标评价算法的性能。

(1)作业完成时间：j_completion。

(2)最大收益：Profit。

(3)平台资源利用率：Utilization。

根据作业的开始时间和执行时间可以得出作业的完成时间，j_start是作业的开始执行时间，j_execute是作业j的全部任务执行完毕所经历的时间，默认全部的作业都同时到达，作业的完成时间j_completion表示为

j_completion=j_start+j_execute

(7)

j_profit为在SLA协议中规定的截止时间之前完成作业j可获得的收益，根据服务商可获得的最大收益，用RP Model中奖惩共存收益函数f(j_profit)衡量

(8)

式(3)中所提到的α，β，γ分别为奖励比率、惩罚比率以及最大惩罚倍数，设置α=0.3，β=0.5，γ=2，a，b分别为时间限定倍数，设置a=b=1.5。

定义平台资源利用率Utilization时，使用作业完成后所占资源与时间的乘积和平台中整体资源与时间的乘积比值表示

(9)

其中，Rtotal是平台中全部资源数量，Ttotal是全部作业执行完毕后的总时间，Rtaski是作业j中第i轮任务执行所占用的资源数量，Ttaski是第i轮任务所需执行时间。

3.3 实验结果与分析

3.3.1 作业执行的高效性

如图2所示，为4个调度器在全部作业的平均完成时间指标上的对比结果。从图中可以看出，FIFO调度器的作业平均完成时间达到了24.3 min，依次降低是EDF调度器22.3 min、Fair调度器20.8 min以及MPCRS 18 min。FIFO调度器的作业平均完成时间最长是由于当一个作业正在执行时会占用集群中的全部资源，不能使其它作业开始执行，故这样就会延长大部分作业的完成时间，导致全部作业的平均完成时间延长。MPCRS与FIFO、EDF以及Fair相比，作业的平均完成时间有所减短。

图2 作业平均完成时间

如图3所示，为每个作业在不同调度器的调度下执行完毕后的完成时间对比结果。首先可以看出3个类型的作业随着数据量的增大，作业完成时间在不断变长；其次在数据量相同时，不同类型的作业完成时间差距很小，说明资源的统一分配对不同类型的作业性能不会造成影响；最后通过比较在4种调度器调度下的作业不同执行性能可以看出，使用MPCRS时的作业完成时间明显短于使用其它3种调度器时的时间，但Job9使用MPCRS时完成时间高于使用EDF和Fair的时间，这是由于在选择Job9最大轮数执行方案时，在最大标准时间范围内，所选的任务轮数最多，这样会使得作业的整体完成时间较轮数较少时有所延长，而且TRN算法和MRNS算法在计算时具有一定的时间消耗，故在作业执行过程中完成时间有一定延长是不可避免的。这个完成时间的延长范围在可允许的范围内，是可以容忍的。

图3 每个作业完成时间

3.3.2 收益最大化

根据上节作业执行的高效性实验中可以得出每个作业的执行性能，但本文设计MPCRS的最终目标是使服务商获取最大收益，故依据式(1)和式(3)，得到如图4所示的使用不同调度器时的最大化收益。其中Ideal为总收益的理想值

图4 服务商收益

(10)

从图4中可以看出使用不同的调度器时，服务商可以获得的最大收益差值较大。其中使用FIFO获得的总收益与理想值Ideal差值最大，使用MPCRS可获得的总收益与理想值Ideal差值最小。由于大部分作业的奖励值仍为0和有部分延迟作业的存在，故使用MPCRS时服务商可获得的总收益值与理想值仍有一定差距。

3.3.3 平台资源利用率的高效性

图5为在各个统计阶段时，使用4个调度器时的平台资源利用率。从图中可以看出，使用FIFO时的平台资源利用率最低，这是由于当一个作业执行时，浪费了其它空闲资源。使用EDF时同样不能充分利用平台资源，使得大量空闲资源浪费。使用Fair调度器时的资源利用率较高，但仍然没有使用MPCRS时的资源利用率高。MPCRS中的MRNS算法考虑了平台资源利用率的影响，故在各个统计阶段中大部分的资源利用率都在90%以上，但在第5阶段和第8阶段资源利用率较低，这是在统计阶段时作业执行情况的影响。

图5 平台资源利用率

4 结束语

结合服务商与用户之间的利益关系，本文提出一种RP Model作为服务商收益最大化的评价标准，并以该评价标准为目标提出了MPCRS，其中具体包括TRN算法和MRNS算法。TRN算法以RP Model中的收益值为标准，根据各个作业的Map和Reduce任务执行时间，确定出作业在不同奖惩阶段的Map和Reduce的最大轮数组合以及最大标准时间；MRNS算法以TRN算法得出的结果为输入，在满足平台资源利用率的前提下，选择出具有局部最大收益的作业和该作业的任务最大轮数方案，从而制定出TS策略。实验结果表明，本文设计的作业调度器MPCRS所生成的TS策略能够最大程度的缩短每个作业的完成时间，提高平台资源利用率，并且在服务商获得最大收益的同时，用户能够得到较为准确的截止时间，真正意义上实现了服务商和用户的利益共赢。