个数据虚拟划分为 n 个区, 在 map 后, 统计每一个区的 key 的数量, 然后考虑 Reduce 负载均衡为这些区分配一个 Reduce 节点文献 [4] 假设节点存在不一样的执行能力, 根据节

2015-10-28 11:21:24 http://www.cnki.net/kcms/detail/51.1196.tp.20151028.1121.104.html 优先出版计算机应用研究第 33 卷 Hadoop 平台中一种 Reduce 负载均衡贪心算法 * 刘朵, 曾锋 *, 陈志刚, 姚亦韬 ( 中南大学软件学院, 长沙 410075) 摘要 :MapReduce 是目前广泛应用的并行计算框架, 是 Hadoop 平台的重要组成部分主要包括 Map 函数和 Reduce 函数 Map 函数输出 key-value 键值对作为 Reduce 的输入, 由于输入的动态性, 不同主机上的 Reduce 处理的输入量存在不均衡性如何解决 Reduce 的负载均衡是优化 MapReduce 的一个重要研究方向本文首先对整体数据进行抽样, 通过适量的样本分析数据, 达到较小的代价获得可靠的 key 分布 ; 然后, 提出贪心算法代替 Hadoop 平台默认的 hash 算法来划分数据, 实现 Reduce 负载均衡本文所提贪心算法主要思想是根据抽样数据, 求取所有 key 频次的和对于 Reduce 节点数量的平均值, 然后依次为每一个 Reduce 分配一个接近平均值的负载, 从而达到整体的负载均衡模拟实验表明, 本文所提算法与默认的 hash 分区算法相比, 运行时间节约 10.6%, 达到更好的负载均衡关键词 :MapReduce; 贪心算法 ;reduce 负载均衡 ; 抽样中图分类号 :TP311 Greedy algorithm for Reduce load balancing on Hadoop platform Liu Duo, Zeng Feng *, Chen Zhigang, Yao Yitao (School of Software, Central South University, Changsha 410075) AbStract: MapReduce is used wildly as a parallel computing framework. mainly including the Map function and Reduce function. Map function has the output of the key-value pairs. which are the input of the Reduce function. As a result. the input of Reduce is dynamic. The load balancing of Reduce hosts has an important impact on the efficiency of MapReduce. In this paper. firstly. the overall data are sampled. The aim is to obtain reliable key distribution at a cheap price. Then. a greedy algorithm is proposed to divide data to achieve Reduce load balancing. taking the place of hash algorithm. The main idea of the greedy algorithm proposed in this paper is to assign the right job to a Reduce host for the best load balancing in each step. Simulation results show that proposed algorithm has better performance than the other two algorithms. Compared with the default hash partitioning algorithm. the proposed algorithm has the running CPU time decreased by 10.6%. and achieves better load balancing. Key Words: MapReduce; Greedy algorithm; load balancing of Reduce; sampling 0 引言大数据的处理目前被广泛的应用和研究,MapReduce 框架是目前应用最为广泛的并行计算框架 [1] Hadoop 是目前应用最为广泛的一个 MapReduce 实现 [2] MapReduce 包括 Map 函数和 Reduce 函数 Map 函数处理输入的 key/value 键值对, 输出中间值的 key/value 键值对, 作为 Reduce 的输出 Map 处理的数据是静态的, 每一个 Map 处理的数据大小是相同的而每一个 Reduce 处理的数据是动态的, 数据量可能不相同, 因此, Reduce 负载存在不均衡性在每个节点的数据处理能力一致的条件下, 任务的执行时间由处理数据量最大的 Reduce 节点确定因此 Reduce 的负载均衡影响 MapReduce 的运行效率目前 Hadoop 平台中 MapReduce 默认使用 hash 算法对数据进行划分该算法并没有考虑到数据中 key 值的分布, 以及每个 key 产生的 hash 值冲突情况例如, 具有相同 hash 值的 key 会被分配到同一 Reduce 节点, 从而导致 Reduce 负载的不均衡可见,hash 算法存在一定的局限性, 没有考虑 Reduce 的负载均衡本文首先通过抽样获取 key 的频次分布, 然后针对抽样数据提出贪心算法实现 Reduce 节点的负载均衡 1 相关工作由于 MapReduce 默认的 hash 算法没有考虑 MapReduce 中间数据的, 从而影响整个 MapReduce 的运行效率为解决这个问题, 国内外学者对 MapReduce 的数据划分的均衡性进行了研究文献 [3] 提出了基于虚拟分区的负载均衡算法该算法把整 ---------------------------- 基金项目 : 国家自然科学基金资助项目 (61103202); 国家教育部博士点基金资助项目 (20110162120046); 中南大学教师研究基金资助项目 (2014JSJJ019) 作者简介 : 刘朵 (1991-), 男, 湖南隆回人, 硕士研究生, 主要研究方向为大数据 ; 曾锋 (1977-), 男 ( 通信作者 ), 副教授, 硕导, 博士, 主要研究方向为无线网络数据挖掘云计算软件工程 ; 陈志刚 (1964-), 男, 教授, 博导, 主要研究方向为是计算机网络与分布式系统 ; 姚亦韬 (1992-). 男, 硕士研究生 ; 主要研究方向为大数据.

个数据虚拟划分为 n 个区, 在 map 后, 统计每一个区的 key 的数量, 然后考虑 Reduce 负载均衡为这些区分配一个 Reduce 节点文献 [4] 假设节点存在不一样的执行能力, 根据节点的执行能力分配 Reduce 负载, 实现 Reduce 的负载均衡文献 [5] 通过抽样后, 寻找分位数的方法来确定数据划分文献 [6] 通过数据抽样后, 将 key 划分为大负载 key, 中负载 key 和小负载 key, 针对不同的 key 分别做不同的处理, 大负载 key 划分成多个子 key, 然后分配到不同的 Reduce 节点上, 将中负载 key 打包后直接分配到 Reduce 节点, 小负载 key 直接使用 hash 划分文献 [7] 提出了 LAB 算法来划分数据, 用一种启发式的方法把特定的 key 对应的数据集分配到最适合的 Reduce 节点上, 然后为下一个 key 寻找分配的 Reduce 节点, 以此类推, 把每个 key 分适合的节点上文献 [8] 提出了 Cluster 组合 Cluster 分割两种算法来对数据进行划分 Cluster 组合应用倾斜度小的中小负载 key, 将 key 按频次排序然后依次划分到 Reduce 上,Cluster 分割应用于倾斜度大的大负载 key, 将大负载 key 切分成多个子 key, 然后依次划分文献 [3,4,7] 提出的算法都需要在 MapReduce 程序运行过程中进行调整, 针对不同的 MapReduce 程序往往需要有不同的调度处理, 操作复杂性高文献 [5] 分位点的 key 通常需要划分到多个 Reduce 上, 处理起来比较复杂, 并且误差较大文献 [6] 对于大中小负载的确认与计算比较复杂文献 [8] 没有考虑由于抽样误差导致未抽取的 key 没有分配的情况基于上述研究工作, 本文尝试完善样本抽样和 Reduce 负载均衡机制首先通过抽样获取 key 的分布, 分析抽样的样本规模和准确度之间的关系, 其次理论分析 hash 算法的不足, 提出贪心算法实现 Reduce 负载均衡, 并通过大规模的数据实验验证算法的有效性 2 算法分析与设计 2.1 问题分析 Hadoop 平台中 MapReduce 默认的划分数据方法是 hash 算法, 根据处理对象的 key 分配 Reduce 主机, 如式 (1) 所示 num = (key.hashcode() & Integer.MAX_VALUE) % nreducetsk (1) 其中 num 是 Reduce 编号,nReduceTsk 是 Reduce 的数量 key.hashcode 为哈希码, 使用的是 BKDR 算法字符串 str 的 hash 码计算如式 (2) 所示 n Hash str[ i]* 31( n i) (2) i 0 从公式 (1) 和公式 (2) 可以看出使用默认的 hash 算法将 key 划分到 Reduce 节点上, 完全取决于 key 的 hash 值, 没有考虑该 key 的其他信息, 这样会存在以下两种情况而使数据产生倾斜 a) 使用 hash 算法时, 多个 key 的 hash 码对 Reduce 节点数量取模之后可能具有相同的值, 从而使数据划分集中于某一个 Reduce 节点, 造成数据不均衡 b)hash 算法没有考虑 key 的频次, 可能存在一些频次大的 key 被划分到同一个 Reduce 节点, 从而造成数据不均衡如图 1 所示, 有 3 个数据节点,Map 端输入数据有 6 个 key 值, 每个 key 值的数据量不相等, 但每个数据节点的数据总量是相等的图中的 key 值 K1 的数据量为 15, 表示为 K1:15 计算可得的数据总量为 75 假设 K1-K6 的 hash 码分别与 1-6 对应, 则由公式 (1)hash 算法分区之后,Reduce 端输入的数据量将不相等, 出现了较大的数据倾斜, 三个 Reduce 节点的数据量分别为 75,111 和 39 K1:15 K2:21 K4:5 S um: 75 K 1:53 K4:22 S um: 75 K5:20 K6:14 K 1:25 K2:29 K3:6 S um: 75 Map K2 : 77 K5:34 S um: 111 K4:7 K5:5 K6:3 Hash 分区 K 1:13 K2:27 K6:1 S um: 75 K3:21 K6:18 S um: 39 K4:10 K5:9 K3: 15 源数据中间数据图 1 中间数据不平衡示例基于上述分析,Reduce 数据划分需要考虑 key 的聚集和频次问题本文拟对数据进行抽样, 获取每个 key 的频次, 求取频次对于 Reduce 节点数量的平均值, 然后依次为每一个 Reduce 分配一个接近平均值的负载, 从而达到整体的负载均衡本文

改进后的 Hadoop 作业运行流程图如图 2 所示 : sampler map() reduce() mapper Reducer 输入 partitioner Partition File HDFS 图 2 改进后的 Hadoop 作业运行流程图图中的阴影部分是新增加的内容, partioner 方法是采用自定义的方法代替系统默认的方法 2.2 抽样抽样 [9] 是从目标数据中抽取一部分样品单位, 基本要求是保证所抽取的样品单位对全部样品具有充分的代表性本文算法是针对海量的数据进行分析, 如果对所有的数据进行统计, 花费代价非常大, 因此采用抽样技术, 获取 key 的频次本文采用系统抽样来对总体进行抽样系统抽样根据样本容量, 首先确定抽选间隔, 然后随机确定起点, 每隔一定的间隔抽取一个单位的一种抽样方式, 是纯随机抽样的变种在系统抽样中, 将总体从 1~N 连续编号, 抽样距离 K=N/n 式中 N 为总体单位总数,n 为样本容量然后, 在 1~K 中抽一随机数 k1, 作为样本的第一个单位, 接着取 k1+k,k1+2k,, 直至抽够 n 个样本为止系统抽样单位在总体中是均匀分布的, 且抽取样本可少于纯随机抽样特点, 能够很好的反映总体情况 [10] 从统计学原理, 我们可以发现随着样本的增大, 抽样的准确度越高设事件 A 发生的概率为 p, 在 n 次重复实验中事件 A 发生次数为 m, 当 n 充分大时 ( 称之为大样本 ), 近似有 m np ~ N(0,1) np(1 p) 取置信区间为 100%, 则有 m np 3 3 np(1 p) np 3 np(1 p) m np 3 np(1 p) 3 np(1 p) m 3 np(1 p) 1 1 np np np 可知误差为 3 np(1 p) 1 1 3 np np n 可见, 随着 n 的增大,m 的误差减少, 抽样的准确度越高 2.3 贪心算法分析 [11] 贪心算法每一步使所做的选择都是当前最佳的, 期望通过局部最优选择来产生出一个全局最优解本文所提贪心算法主要思想是根据抽样数据, 求取所有 key 频次对于 Reduce 节点数量的平均值, 然后依次为每一个 Reduce 分配一个接近平均值的负载, 从而达到整体的负载均衡具体算法如下 : avg Step 1: 将 key 按频次, 从小到大排序 Step 2: 计算所有 key 的频次和对于 Reduce 数量的均值 Step 3: 将频次大于均值 avg 的 key 拆分分配到 1 个负载为 0 的 reduce 节点上, 记录 key 和 Reduce 分配的对应关系, 将 key 的频次减去 avg 重复处理, 直到 key 的频次小于 avg 若 key 的频次不为 0, 则将 key 重新按序插入队列 Step 4: 重复 Step 3, 将所有的频次大于均值 avg 的 key 处理完毕 Step 5: 选择 Step 3 没有涉及的 Reduce 节点从后至前遍历队列, 求取 key 的频次与该节点当前负载的和, 如果该和不超过 avg, 则将该和作为 Reduce 节点的当前负载, 记录 key 和分配的 Reduce 的对应信息, 删除队列该 key 的信息重复上述操作, 直到遍历完整个队列 Step 6: 重复步骤 5, 均衡余下 Reduce 节点的负载, 记录

key 和分配的 Reduce 的对应信息 Step 7: 对输入的每一个 key, 通过步骤 3 4 5 6 产生的记录, 返回该 key 分配的 Reduce 的编号图 2 的例子通过贪心分区算法进行数据划分之后的结果如图 3 所示可以看出本文所提贪心分区算法获得较好的 Reduce 负载均衡, 优于默认分区算法 K1:15 K2:21 K4:5 K5:20 K6:14 k2:75 K 1:25 K2:29 K3:6 K4:7 K5:5 K6:3 Map 贪心分区算法 K1:53, k4:22 K1:13 K2:27 K5:9 K3:15 K4:10 K6:1 K5:34, k3:21, K6:18, K2:2 源数据中间数据图 3 中间数据平衡后示例 2.4 基于贪心算法的数据划分算法实现通过抽样统计获得所有 key 频次, 对于在抽样过程中, 没有抽取出的 key, 认为是小概率数据不影响 Reduce 的负载均衡没有抽取到的 key, 使用 hash 分区算法, 划分到对应的 Reduce 上定义全局变量 KeyReduce,KeyReduce 存储的是 Key 和应该划分的 Reduce 编号的对应关系当 key.freq>avg 时记 key 为大负载 key 本文设计的一种贪心数据分区算法的的伪代码如下 : 1 keyreduce GreadyPart() 2 { 3 读取 PartionFile 文件中的 key 的频次, 记录到键值对序列 KFreq 中 ; 4 SortByFreq(KFreq);.// 将 KFreq 按照频次从小到大排序 5 avgvalue = sum/m;// 求每个 reduce 平均处理的 key 频次综合 6 reduce = 0; 7 // 将大于 key 平均频次的所有 key 的拆分 8 for(int i=0;i<kfreq.size;i++) 9 { 10 if(kfreq[i].value > avg) 11 { 12 while(kfreq.value > avg) 13 { 14 KeyReduce.Add(key,reduce); 15 KFreq[i].Value -= avgvalue 16 } 17 KFreq.Sort();// 将改变后的元素, 调整到合适的位置 18 } 19 else 20 { 21 break; 22 } 23 } 24 for(int i=0;i<m;i++) 25 { 26 reducesum = 0;// 每个 Reduce 分配的 key 的总和 27 for(int j=kfreq.size()-1;j>=0;j--) 28 { 29 int value = KFreq[j].value; 30 if(avgvalue>=reducesum+value) 31 { 32 reducesum += value; 33 KeyReduce.add(KFreq[j].key, i);

34 KFreq.remove(j); 35 } 36 } 37 } 38 return KeyReduce; 39 } 40 public int getpartition(text key, Text value, int numpartions) 41 { 42 if(kyereduce == null) 43 { 44 GreadyPart(); 45 } 46 if(keyreduce[key]!= null) 47 { 48 return KeyReduce[key].Reduce; 49 } 50 return (key.hashcode()& Integer.MAX_VALUE) % numreducetasks; 51 } 算法描述 : 按照 key 的频次从小到大排序, 求出所有 key 的频次之和, 计算划分到 m 个 Reduce 上的平均值 avg 如果某 key 的频次大于 avg, 则需要将其划分到多个 Reduce 上, 该 key 称作大负载 key 代码 8-23 行是将大负载 key 划分到不同的 Reduce 上, 并将 key-reduce 的对应关系加入 KeyReduce 中承载了大负载 key 的 Reduce 不再划分数据代码 24-37 行是为没有承载大负载 key 的 Reduce 节点均衡负载依次为每一个 Reduce 节点分配 key 对于第 i 个 Reduce 节点, 选择当前队列中频次最大的 key, 如果满足分配该 key 后, 第 i 个 Reduce 节点的负载不大于 avg, 则将 key 分配到第代码 40-51 行是 getpartition(), 通过该方法返回对应 key 划分的 Reduce 的编号代码 46-49 行, 如果 key 在 KeyReduce 中有对应的 Reduce 的编号, 直接返回该编号, 否则认为 key 在抽样过程中没有抽到, 是小数据, 不影响负载均衡, 使用默认的分区算法划分到对应的 Reduce 节点上如果 key 为大负载, 在 KeyReduce 中能够找到多个值, 可将该 key 按照比例分配到各个 Reduce 节点上 3 实验结果及分析 3.1 硬件平台及部署本文的实验集群由 7 台计算机组成, 每台计算机有 2G 内存,300G 磁盘空间包括 1 个主节点 :master.csu 和 6 个工作节点 :slave1.cus-slave6.csu, 节点的部署信息如表 1 所示网络环境 : 校园内部局域网, 操作系统 :Centos 6.6,Java 环境 : JDK1.6,Hadoop 版本 :Hadoop-1.2.1, 开发工具 :MyEclipse8.6 3.2 实验结果及分析以 WordCount 为实例, 进行实验分别比较默认的 hash 分区算法, 分位数分区算法和贪心分区算法从运行时间和 Reduce 的负载均衡两个角度比较三种算法的优劣数据从网上随机下载, 分别比较数据集在 432M,4.32G,8.64G,20G 情况下的运行时间以及在数据集在 4.32G 时各个节点的负载均衡在数据量比较小的情况下, 如在数据量为 400M 左右时, 两种算法的执行时间相当但随着数据量的增大, 使用贪心分区算法的效率会越来越高, 在数据达到 20G 的时候, 与默认的 hash 分区算法相比, 贪心分区算法降低执行时间约 10.6%, 实验数据如图 4 所示在实验数据量为 4.32G 时, 每个 Reduce 节点的负载情况如图 5 所示, 贪心分区算法的每个 Reduce 负载基本相同, 而使用默认的 hash 分区算法的每个 Reduce 的负载有较大的数据起伏分位数数分区算法的均衡度好于 hash 分区算法, 但比贪心算法差与 hash 分区算法和分位数数分区算法相比, 贪心算法的负载均衡度提高分别为 44.4% 和 9.2% 由此可见, 在均衡衡负载和时间效率这两个方面, 贪心分区算法要优于 hash 分区算法 i 个 Reduce 节点上直至遍历完整个队列表 1 节点部署情况服务器 IP 服务器主机名功能 192.168.1.120 master.csu 主节点 (namenode 和 jobtracker) 192.168.1.121 slave1.csu 从节点 1(DataNode 和 TaskTracker) 192.168.1.122 slave2.csu 从节点 2(DataNode 和 TaskTracker) 192.168.1.123 slave3.csu 从节点 3(DataNode 和 TaskTracker) 192.168.1.124 slave4.csu 从节点 4(DataNode 和 TaskTracker) 192.168.1.125 slave5.csu 从节点 5(DataNode 和 TaskTracker) 192.168.1.126 slave6.csu 从节点 6(DataNode 和 TaskTracker)

2500 2000 1500 1000 500 0 Hash 分区算法分位数数据划分算法贪心分区算法 432MB 2G 4.32G 8.64G 20G 18000000 16000000 14000000 12000000 10000000 80000000 60000000 40000000 20000000 0 Hash 分区算法分位数分区算法贪心分区算法图 4 大小不同的数据集的执行时间比较图 5 同一数据集 (4.32G) 不同节点上的负载比较 4 结束语本文主要是对 MapReduce 的中间数据平衡进行研究 Reduce 函数使用 Map 函数产生的中间结果作为输入数据, 是动态的数据 MapReduce 默认使用 hash 算法来进行数据划分, 每个 Reduce 节点的负载不平衡本文通过抽样获取 key 的频次, 使用贪心算法代替 hash 算法, 均衡 Reduce 的负载无论是理论分析还是实验验证, 均表明贪心分区算法是一个良好的数据分区算法 [10] 于寅, 等. 高等工程数学 [M]. 武汉 : 华中科技大学出版社, 2012: 340-355 [11] Cormen T H, LeiserSon C E, et al. 算法导论 [M]. 北京 : 机械工业出版社, 2006: 222-239 参考文献 : [1] Dean J, Ghemawat S. MapReduce: simplified data processing on large clusters[j]. Communications of the ACM, 2008, 51(1): 107-113. [2] White T. Hadoop: the definitive guide: the definitive guide[m]. [S. l. ]: O'Reilly Media, Inc. ", 2009. [3] Fan Y, Wu W, Cao H, et al. LBVP: A load balance algorithm based on Virtual Partition in Hadoop cluster[c]//proc of IEEE Asia Pacific Conference on Cloud Computing Congress. 2012: 37-41. [4] Gao Z, Liu D, Yang Y, et al. A load balance algorithm based on nodes performance in Hadoop cluster[c]//proc of the 16th Asia-Pacific Network Operations and Management Symposium. 2014: 1-4. [5] 韩蕾, 孙徐湛, 吴志川, 等. MapReduce 上基于抽样的数据划分最优化研究 [J]. 计算机研究与发展, 2013, S2: 77-84. [6] Ramakrishnan S R, Swart Gt, Urmanov A. Balancing reducer skew in MapReduce workloads using progressive sampling[c] //Proc of the 3rd ACM Symp on Cloud Computing. New York: ACM, 2012 [7] 余基映. MapReduce 模型的数据分配策略研究 [D]. 武汉 : 华中科技大学, 2013. [8] 耿玉娇. MapReduce 中基于抽样技术的倾斜问题研究 [D]. 大连 : 大连海事大学, 2013. [9] 宛婉, 周国祥. Hadoop 平台的海量数据并行随机抽样 [J]. 计算机工程与应用, 2014, 40(20): 115-118.

个 数 据 虚 拟 划 分 为 n 个 区, 在 map 后, 统 计 每 一 个 区 的 key 的 数 量, 然 后 考 虑 Reduce 负 载 均 衡 为 这 些 区 分 配 一 个 Reduce 节 点 文 献 [4] 假 设 节 点 存 在 不 一 样 的 执 行 能 力, 根 据 节

个数据虚拟划分为 n 个区, 在 map 后, 统计每一个区的 key 的数量, 然后考虑 Reduce 负载均衡为这些区分配一个 Reduce 节点文献 [4] 假设节点存在不一样的执行能力, 根据节