结合节点计算能力的MapReduce负载均衡方法

胡林发,付晓东,2,刘 骊,刘利军

(1.昆明理工大学 信息工程与自动化学院,昆明 650500;2.昆明理工大学 云南省计算机技术应用重点实验室,昆明 650500)

0 引 言

随着互联网的迅猛发展,尤其是移动互联网的应用和大数据的普及,数据量迎来爆炸式增长。MapReduce作为一种分布式计算模型,被广泛应用在大规模数据并行计算过程中[1-2]。目前针对MapReduce有多种实现,如Apache的开源Hadoop框架,将MapReduce和HDFS分布式文件系统进行完美融合,深受工业界和学术界的青睐[3]。

数据均衡划分是MapReduce框架在Shuffle阶段需要解决的一个重要问题。用户提交的作业(Job)由一系列分别运行在多台机器上的Map任务(Mapper)和Reduce任务(Reducer)处理完成,作业的完成时间由运行最慢的Reducer决定[4-5]。Hadoop系统默认采用Hash分区方法,即仅根据关键字的哈希值和分区个数确定关键字的分区,这种划分方法虽然可以保证每个分区里不同关键字种类数大致相等,但每种关键字携带的数据记录条数不一定相等,这会使各分区数据总量大小相差悬殊,从而导致各节点负载不均衡问题。研究表明,采用默认的Hash分区方法,超过90%的作业任务出现了Reducer负载不均衡情况,运行时间高出正常任务的22%-38%[6]。先完成任务的节点需要等待滞后节点任务全部完成才能结束当前作业,若中间数据过于集中在某部分Reducer任务节点,先完成任务的节点必须等待其他节点,这个过程会造成集群资源浪费,延长整体作业完成时间,甚至某些节点因资源不足导致任务中断,使作业无法继续推进,从而带来不好的用户体验[7-8]。

文献[9-10]指出,将Mapper产生的中间数据最优地划分到不同分区,使各分区负载均衡是一个NP-Hard问题。针对MapReduce框架中存在的负载均衡问题,目前已有两阶段分区[11]、多阶段分区[12-13]、数据采样分区[14-15]、延迟分区[16]、迁移分区[17]等方法,这些方法将集群中各节点看作是计算能力相同的节点,但在实际数据处理过程中不同代的硬件环境会使每个节点计算能力不相同[18],且不同计算节点的性能差异会影响整个系统的计算效率[19]。在异构环境中,即使所有分区得到相同规模的数据,也会因节点处理能力不同导致Reduce任务完成时间产生巨大差异,存在先完成任务的节点等待滞后节点的问题,作业执行时间因此会被延长,集群中部分计算资源会被闲置,从而降低了作业处理效率,浪费了计算资源。

本文提出一种结合节点计算能力的划分方法,即在数据划分时结合节点计算能力,使各节点数据负载与节点自身的计算能力相匹配,并使大量数据在节点本地处理,降低网络传输时延,从而提升作业的处理效率。本文的主要贡献包括以下3个方面。

1)提出在异构环境中使用Reservoir算法对Map任务产生的中间数据进行抽样,记录样本中关键字的位置和频次,并以此建立关键字分布矩阵。

2)提出一种结合节点计算能力的分区划分方法。在制定分区计划时,本文先采用贪心策略对关键字进行初步分区,使各关键字划分到其频次最高的节点对应分区,然后结合节点计算能力并考虑节点位置关系对初步划分结果进行调整,使各分区负载均衡。

3)设计了一种均衡性衡量方法,该方法综合考虑了数据量和节点的计算能力值,有利于更加全面地衡量分区结果的均衡性。

1 异构环境中负载均衡问题

在MapReduce处理作业时,作业任务分为Map和Reduce任务。执行任务的函数均由用户根据业务需求自定义。将集群中节点数记为r,节点集合为N={n1,n2,…,nr},分区集合为P={p1,p2,…,pr},其中pj(j=1,2…,r)为一个分区。为方便集合节点计算能力划分,这里将pj分区里所有数据作为节点nj上Reduce任务的输入。Map任务产生的中间数据为键值对形式,分区算法会将中间数据按照关键字划分到不同分区。由Reduce任务输入限制知,相同关键字的数据只能被同一个Reduce任务处理,即∀i≠j且i,j=1,2,…,r,pi∩pj=∅。

分区集合里每个分区的实际数据量大小为S={s1,s2,…,sr},令C={c1,c2,…,cr}表示每个节点的计算能力值,τ(τ>0)为可设定的阈值,当满足

(1)

时,集群节点负载不均衡。

(2)

将分区划分方法记为Π(x),则在Shuffle阶段关键字kt会根据Π(x)计算得到分区pj←Π(kt),j=1,2,…,r,然后关键字kt会被划分到分区pj中。此时,其他节点上关键字为kt的数据需要通过网络传输到nj(j=1,2,…,r),则关键字kt需要在网络中传输的数据总量为

(3)

关键字K={k1,k2,…,kΩ}中所有关键字在经过分区方法Π(x)划分之后,共需在网络中传输的数据总量用VTRS表示,则

(pj=Π(kt),j=1,2,…,r)

(4)

(5)

VTRS值大小取决于VLocality值大小,当VLocality越大时,VTRS越小,需要在网络中传输的数据就会越少。

在执行Reduce任务时,节点nj(j=1,2,…,r)上处理pj分区里的数据,将pj数据总量记为sj,用FoS(factor of skew)值SFoS衡量节点负载的均衡性,计算式为

(6)

因此,本文解决的负载均衡问题为结合节点的计算能力值,寻找一种分区划分方法Π(x),使SFoS尽可能小,让各节点负载接近,同时降低网络开销,从而提高作业的处理效率。

2 结合节点计算能力的负载均衡方法

针对异构集群环境中负载不均衡问题,本文提出结合节点计算能力的分区方法LBCC(load balancing in MapReduce combined with computing capacity)。在节点加入到集群时,各节点上运行测试程序,执行默认计算任务。将测试数据集的大小记为V,在节点nj上任务完成所需时间记为Tj,则可得出节点nj计算能力值cj=V/Tj,节点计算能力值集合记为C={c1,c2,…,cr}。在搭建Hadoop环境时,利用配置文件core.xml配置节点的所属机架信息,方便后续利用节点机架信息调整节点负载。

为使各节点负载均衡并降低Shuffle过程中网络通信开销,本文在执行用户提交的计算作业之前,先运行一个抽样作业进行数据抽样,并统计样本数据里关键字的位置和频次分布,由此得到关键字分布矩阵M,然后结合M和节点计算能力值,经过位置划分筛选高低分区以及分区调整等步骤制定分区计划并将其写进缓存文件fcache。分区计划是计算作业分区划分的依据,使计算作业任务运行时各节点负载均衡,从而提高集群资源的利用率和作业执行效率。

2.1 运行抽样作业

在抽样作业Map阶段采用Reservoir抽样算法对数据集进行抽样,然后在Reduce阶段汇总各节点样本数据,依据样本里的关键字位置和频次信息建立关键字分布矩阵,并根据分布矩阵和节点计算能力信息制定分区计划。

在抽样作业Map任务阶段,首先初始化一个关键字集合KL,再按行读取数据集并将数据集中的关键字逐一添加进KL中,具体过程如算法1所示。

算法1对数据集中关键字进行抽样

输入:数据集分片β,样本容量α。

输出:关键字样本集合KL.

1.KL←∅;

2.cnt←0;

3.forlineinβdo

4.k←getKey(line);

5.cnt++;

6.ifKL.size()<αthen

7.KL.add(k);

8.else

9.t←random.nextInt(0,cnt);

10.ift<αthen

11.KL.replace(t,k);

12.end if

13.end if

14.end for

15.outputKL;

当VLocality取最大值时,VTRS取得最小值,需要在网络中传输的数据量最少。显然,对于任意kt,若pj←Π(kt)且满足

(7)

时,VLocality可以取最大值。这里先采用贪心方法,逐一将K={k1,k2,…,kΩ}里关键字分配到可以使VLocality值取最大值的分区,由此得到初次划分结果,然后在此基础上将各分区里关键字进行调整,使各分区负载接近分区期望值,具体过程如算法2所示。

算法2制定分区计划

输入:二维矩阵M=[mtj]Ω×r,C={c1,c2,…,cr}.

输出:分区计划P={p1,p2,…,pr}.

1.pj←∅(j=1,2,…,r);

2.fort←1 toΩdo

3.mtj←max{mt1,mt2,…,mtr};

4.j←getNodeIndex(mtj);

5.pj←pj.add(kt);

6.end for

8.PH←∅,PL←∅;

9.forj←1 tordo

12.ifsj>ejthen

13.PH←PH∪{pj};

14.else

15.PL←PL∪{pj};

16.end if

17.end for

18.forh←1 toPH.size() do

19.pi←PH.get(h);

20.forktinpido

21.pj←getMinNearPartition(PL,pi);

22.pj.add(kt);

24.PL.remove(pj);

25.end if

26.pi.remove(kt);

28.break;

29.end if

30.end for

31.end for

32.outputP={p1,p2,…,pr}.

算法2中,第2—第6行表示依次将关键字kt划分到kt频次最大的节点对应分区上, 直到所有关键字划分完毕,得到初步划分结果P={p1,p2,…,pr}。此时,VLocality取最大值,需要在网络中传输的数据量最小。然而,此时并没有考虑节点负载均衡性,还需要对初步分区计划进行调整。对于分区pj(j=1,2,…,r),sj为pj分区里数据总量,每个节点负载期望值用ej表示,ej表达式为

(8)

算法2中第7—第17行表示将分区按照实际负载是否高于均衡值划分为高分区和低分区,若分区sj>ej则将pj加入高分区集合PH,否则加低分区集合PL。逐渐从高分区里移出关键字数据,当高分区的实际总数据量低于期望值时则停止移出。第18—第30行表示收集高分区里的关键字,并且将从高分区移出的关键字逐一分配给PL中的低分区,从而使低分区数据量逐渐接近期望值。若在关键字调整的过程中,当某低分区实际数据量高于均衡值时,则将该低分区移出集合PL。

算法2中getMinNearPartition方法是为了在集合PL中寻找离分区pi最近的节点分区,首先在PL中寻找与pi同一机架且分区负载最小的分区,若存在直接返回,不存在则在其他机架上寻找,具体过程如算法3。

算法3getMinNearPartition方法实现

输入:低分区集合PL,分区pi.

输出:PL中离pi最近的分区p.

1.p←null;

2.flag←0;

3.forpjinPLdo

4.ifrack(pi)=rack(pj) then

5.ifp!=null&&getLoad(p)>getLoad(pj)

||p=nullthen

6.p←pj;

7.flag←1;

8.end if

9.end if

10.end for

11.ifflag=1 then

12.returnp;

13.end if

14.returnminLoad(PL);

算法3中第3—第10行表示在集合中寻找与分区pi关联节点ni同一机架的其他节点对应分区,其中,第4行rack方法的作用是获取分区的机架位置,第5行getLoad方法表示求取分区的实际负载大小,分区pj的负载计算方法可表示为

(9)

算法3中第11—第14行表示如果在PL中找到了合适分区就直接返回,若没有找到则利用minLoad方法求取整个PL集合中负载最小的分区作为返回结果。

算法2初步划分过程中,需要遍历分布矩阵M中所有行,时间复杂度为O(Ω·r)。在分区筛选过程中遍历分区集合P={p1,p2,…,pr},时间复杂度为O(r)。在分区调整时,需要先将高分区进行排序,这个过程时间复杂度取决于采用的排序算法,本文采用快速排序,所以时间复杂度为O(ΩlogΩ)。在将高分区里部分关键字调整到低分区时,需要通过算法3寻找合适分区,最好的情况下只有少量关键字需要进行调整,时间复杂度为O(r),而在最坏的情况下,大量关键字需要进行调整,此时时间复杂度为O(Ω·r)。综上,算法2的时间复杂度为O(ΩlogΩ)。

整个抽样作业的输出分区计划为P={p1,p2,…,pr}。为方便计算作业利用分区计划进行分区,这里将其转化为以关键字kt为键、以kt所属分区编号为值的键值对形式,并将其写入到缓存文件。

2.2 执行计算作业任务

计算作业以全量数据为输入,并按照制定的计划进行分区。在Mapper阶段读取缓存文件fcache,并将其转化为以关键字kt为键、分区编号为值的键值对结构,将其记为F。分区方法Π(kt)首先在F中查找是否存在关键字kt,若存在则直接输出分区pj(j=1,2,…,r),否则按照Hash方法得到pj,分区流程如图1所示。

图1 分区划分流程图Fig.1 Partition flow chart

由于在抽样过程中,可能存在少量频率较小的关键字可能没有被抽样到,所以在这里使用Hash方法作为辅助方法。任意关键字kt根据Π(kt)计算得到分区pj后,将关键字kt与其携带的数据写入到pj分区文件中。在计算作业Reduce阶段各计算节点分别从Map任务节点拉取属于本节点的中间数据分区文件,并运行Reduce任务,直至任务运行结束,输出作业的计算结果。

3 实 验

本文LBCC方法在分区划分时考虑各节点计算能力的同时,对网络传输开销进行了优化。为验证本文方法效果,采用NoLFA方法[20]、SBaSC方法[21]和DEFH(default Hash)方法对比。NoLFA方法基于LEEN思想并结合了节点计算能力的差异性,适用于异构集群环境,但其与本文方法相比有以下几点不同:①NoLFA方法直接在计算作业执行过程中通过主节点获取关键字频次信息,这增加了主节点负担,降低了集群元数据处理效率,本文使用抽样作业得到关键字频次分布信息,避免了元数据处理效率降低问题;②分区计划制定时,NoLFA方法直接按照LEEN方法思想进行处理,而本文方法在做了一次初步划分之后,对低分区进行调整,可以快速使最低分区总数量达到均衡值,而且本文方法分区均衡性比NoLFA方法更好;③本文方法在调整分区负载过程中同时考虑了节点计算能力和节点位置差异,能更好地适应异构集群环境。SBaSC方法使用了贪心方法思想划分分区,达到了均衡各节点负载的目的并且提升了作业计算效率,但其将集群中所有节点看作相同的计算能力,忽略了各节点处理能力的差异性。

3.1 数据集

为测试倾斜度对算法性能的影响,实验采用人工数据集和2个公开数据集。人工数据集是使用程序生成不同倾斜率的数据集[22],实验时将人工数据集上传到HDFS系统,并让其分散在不同节点上进行存储,每组实验均基于该数据集执行单词统计任务。2个公开数据集分别为维基百科数据集[2]和社交网络数据集LiveJournal[21]。维基百科数据集包含了大量的文本数据信息,实验时在对该数据集进行预处理后将其作为单词统计作业的输入。LiveJournal数据集中包含了大约1亿个用户社交网络数据,实验中使用该数据集作为关联用户数目统计作业的输入。

3.2 实验环境

实验采用物理节点与虚拟节点结合的方式,模拟异构集群环境中不同计算能力节点环境。每组实验涉及到的物理机节点配置为I3、8核、16 GByte内存、500 GByte磁盘空间,虚拟机节点在I5机器上搭建,单个虚拟节点分配4核、8 GByte内存、100 GByte磁盘空间,Hadoop版本为2.10,所有节点均采用CentOS 6.9系统,物理机节点和虚拟节点个数分别由实验需求确定。

3.3 数据倾斜程度对性能影响

通常关键字频次服从Zipfian分布[4],在关键字列表K={k1,k2,…,kΩ}中,排在λ(λ=1,2,…,Ω)位置的关键字出现频率f(λ)可以表示为

(10)

(10)式中,z≥0为倾斜程度控制参数,z值越大则表示数据集中关键字的频次分布越集中,当z=0时表示关键字频率相同,即所有关键字频次均匀分布。

分别设置人工生成不同倾斜率数据集倾斜度z=0.2、z=0.4、z=0.6、z=0.8、z=1.0五组实验,实验前准备一个关键字个数为20 000的单词列表,各关键字根据(10)式得到关键字频率f(λ)。在向输出文件里写数据时,f(λ)作为关键字kλ写入的概率,以此生成包含10亿单词的数据文件。

配置不同分区算法并提交单词统计作业。搭建包含2个物理节点和3个虚拟节点的Hadoop集群环境,通过文件配置使每个节点既是DataNode节点又是NodaManager节点。实验结果汇总信息如表1、图2—图3所示。

表1 在不同倾斜度下的FoS值Tab.1 FoS value at different skew degree

表1展示的是在不同倾斜度数据集作为输入的情况下各种算法得到的FoS值(表1中,除倾斜度外的数值为实际数值乘以10-5)。不难发现,在各种倾斜度下,本文LBCC方法FoS值最小,即均衡性表现最好,可以使各节点负载更加均衡。DEFH方法在各种倾斜度下FoS值都最大,均衡性最差,这样会导致集群汇中部分节点负载远高于其他节点,从而降低作业的执行效率。NoLFA方法FoS值比LBCC方法高,最大可以是LBCC方法的236.2倍,说明此时NoLFA分区结果节点负载均衡程度远不如本文LBCC方法。

图2 不同倾斜度下本地化率值Fig.2 Locality value at different skew degree

图3 不同倾斜度下执行时间Fig.3 Execution time at different skew degree

由图2可见,在相同数据量下,随着关键字频次倾斜度的增加,Locality值呈下降趋势,即需要在网络中传输的数据量逐渐增加。DEFH方法、SBaSC方法变化幅度比较小,因其没有考虑网络开销优化,这2种方法需要在网络中传输的数据量占总数据量20%左右。当倾斜率较高时,LBCC方法Locality值会比NoLFA稍低,这是由于在数据倾斜率较高时部分关键字在各节点分布不均匀,本文方法在调整分区过程中,为了使各分区均衡性更好,会结合节点计算能力和节点位置信息将部分关键字调整到最低分区,会牺牲一部分Locality值,相对于NoLFA方法在倾斜率较高时Locality值会存在一定差距,但比其他方法本地化率更高。

由图3可见,在数据总量相同且节点个数一定情况下,作业执行时间也随之增大。LBCC方法结合节点计算能力将中间数据更加均衡地划分,缩短了整体作业的完成时间。LBCC方法相较于NoLFA、SBaSC、DEFH方法在效率上都有较大提高。

3.4 节点个数对性能影响

为测试异构环境下节点个数对算法性能的影响,实验环境初始设置为1个物理节点和2个虚拟节点共3个节点,之后每组实验在此基础上依次增加1个物理节点和1个虚拟节点,节点个数依次设置为3、5、7、9、11个。本次实验分别采用维基百科数据集和社交网络LiveJournal数据集,每次配置好环境后,将数据集上传到HDFS文件系统,数据文件会以块的形式分散存储在集群中各节点上。实验中每种算法运行多次后求取各项指标平均值,汇总信息如表2—表3、图4—图7所示。

表2 在维基百科数据集上FoS值Tab.2 FoS value for different number of data nodes on Wikipedia dataset

表3 在LiveJournal数据集上FoS值Tab.3 FoS value for different number of data nodes on LiveJournal dataset

由表2—表3可知,在不同节点个数实验环境中,无论采用哪种数据集作为作业的输入,本文LBCC方法分区结果的FoS值最低,即均衡性表现最好,可以使各节点负载更加均衡。本文LBCC方法在调整各节点负载时考虑了节点计算能力,让各分区之间的负载与节点自身计算能力相匹配,与其他各节点负载相均衡。 DEFH方法根据关键字哈希值进行划分,并没有考虑各节点的负载均衡性,所以FoS值比较大。另外在异构环境中,SBaSC没有考虑节点的计算能力,所以导致各节点负载差异也很大。

图4 在维基百科数据集上不同节点个数下的本地化率Fig.4 Locality value for different number of nodes on Wikipedia dataset

图5 在LiveJournal数据集上不同节点个数下的 本地化率Fig.5 Locality value for different number of nodes on LiveJournal dataset

由图4—图5可见,随着节点的增加,Locality值总体上呈下降趋势。文献[5]指出,相同关键字的频次在集群中各节点均匀分布时,数据本地化率取决于节点的个数,即VLocality=1/r,在数据值上将与公式计算的结果相等。由此可知,随着节点个数的增加,Locality值会随之下降。在不同节点个数环境下,NoLFA方法和LBCC方法的分区结果中本地化率比较接近,SBaSC和DEFH方法由于没有考虑网络开销优化,Locality值比较低,需要在网络中传输的数据量比较大。数据集LiveJournal上关键字的频次比较集中,大量的关键字携带的数据可以在节点本地进行处理,不需要通过网络传输到其他节点,所以通过LBCC和NoLFA方法得到的Locality值比较高。

由图6—图7可见,随着节点个数的增加,任务完成时间逐渐降低。另外,在每组实验中,本文LBCC方法在效率上优于其他分区方法。图7中NoLFA和LBCC方法相较于图6差别较大,这是由于使用NoLFA和LBCC方法可以使社交网络LiveJournal数据集绝大部分在节点本地处理,另外在异构环境中,考虑了节点负载均衡性,使各节点的负载与节点计算能力相匹配。在维基百科数据集上,LBCC方法在作业运行效率上比NoLFA方法提高7.0～15.4百分点,比SBaSC方法提高17.9～23.1百分点,比DEFH方法提高11.0～30.8百分点。在社交网络LiveJournal数据集上,LBCC方法在效率上比NoLFA方法提高2.8～7.6百分点,比SBaSC方法提高8.1～15.4百分点,比DEFH方法提高10.1～15.9百分点。

图6 在维基百科数据集上不同数据节点下 任务完成时间Fig.6 Execution time for different nodes on the Wikipedia dataset

图7 在LiveJournal数据集上不同数据节点下 任务完成时间Fig.7 Execution time for different number of nodes on LiveJournal dataset

4 结束语

本文提出通过Reservoir抽样方法获取Map产生的中间数据分布信息,然后结合节点计算能力解决MapReduce在分区过程中的负载均衡问题。实验结果表明,本文方法得到的分区结果会使各节点负载更为均衡,提高了作业处理效率,同时优化了网络传输代价。本文方法在集群异构的环境中具有良好的性能优势,计算效率相对于现有分区方法有显著提升,为MapReduce计算模型负载均衡提供了一种更加高效的解决方案。