胡忠奎，屈波，黄斌，黎文阳

（1.四川大学计算机学院，成都　610065；2.中国人民解放军78098部队，成都　611200）

一种基于虚拟处理区间划分的负载均衡等值连接算法

胡忠奎1，屈波2，黄斌1，黎文阳1

（1.四川大学计算机学院，成都610065；2.中国人民解放军78098部队，成都611200）

0　引言

以 Spark为新兴代表的内存计算，在保持了MapReduce优良传统的同时，大幅提升了其运行效率，尤其在迭代式计算与交互式数据分析方面表现优异，逐渐受到业界的青睐。Spark SQL是Spark生态系统的关键组成部分，它实现了在Spark系统上对关系型数据的查询，是对Shark和Hive数据仓库系统的升级。连接操作，特别是等值连接操作，是数据分析中常见且代价很高操作之一，Spark SQL由于采用了Hash划分技术来处理等值连接操作：根据连接属性上Key的Hash值确定Key对应数据的所属分区，每个分区由一个Reduce处理单元执行连接操作。由于现实中数据在分布上符合帕累托法则（80/20定律）即存在不均衡性，这就导致在数据等值连接查询中，由于连接属性数据分布的不均衡，造成不同Hash值对应的数据量的不均衡，出现数据的聚集问题，从而导致某些计算节点的负载过重，极大地降低了大数据查询分析的性能，成为当前我们必须解决的新问题。

1　相关研究

基于Spark/MapReduce的大数据分析中，常用的连接算法有Simi-Join、Broadcast Join、Repartition Join等[1]。其中Simi-Join和Broadcast Join局限性较大，通常性能较差，Repartition Join适用性最好，在绝大多数情况下具有最好的性能。但Repartition Join在Spark上具有如下的缺点。①需要在查询时对数据进行动态的重划分，通信量较大，尤其在宽表的情况下；②通过Hash函数划分到同一节点的很多事实表元组在外键上具有相同值，内存和计算资源消耗较大，同时容易出现数据倾斜问题。关于连接中的数据倾斜问题，有很多学者进行了深入的研究，并给出解决方案。卞昊穹等人[2]通过预等值连接得到Hash后的数值分布，并对其重新划分得到负载均衡等值连接映射，最后根据映射关系组装等值连接；翟红敏等人[3]基于对等值连接键值的统计信息，进行范围分割数据量，从而达到总体负载均衡的目的；吴磊[4]基于分而治之的思想将倾斜数据和非倾斜数据区别对待，结合了传统等值连接算法、广播等值连接算法等算法思想，解决数据倾斜情况下任务负载分布不均匀的问题；Chen Qi等人[5]通过提出一种新的采样方法较为准确地估计连接属性的数据分布情况，并基于等值连接代价模型计算代价最小的数据分配方案，实现等值连接计算的负载均衡。

2　代价模型和相关分析

定义1连接数据与连接噪声：在数据集的连接属性中可能包含一些共有的连接键值，这些键值对应的数据称为连接数据，而非共有部分的键值对应的数据称为连接噪声。

定义2虚拟处理区间：在实际处理区间（reduce任务）之上，虚拟划分出更多的区间，以解决采样估计误差造成的不均衡问题，达到数据均衡分配的目的。

假设v表示虚拟处理区间的数量，H={H1，H2，…，Hv}表示虚拟处理区间集合。{（K1，C1），（K2，C2），…，（Kn，Cn）}表示采样得到统计数据，其中，Ki表示连接属性上的键值，Ci表示与Ki对应的元组数量，在没有特别说明时，我们约定Ki＜Ki+1成立。实际处理区间的数量为k，并由R={R1，R2，…，Rk}表示。虚拟处理区间到实际处理区间的交叉映射关系可以表示为MapPartition{（H1，Hk+1，…，H（α-1）k+1→R1），（H2，Hk+2，…，H（α-2）k+2→R2），…，（Hk，H2k，…，Hαk→Rk）}。其中，α=v/k。

定义3单个处理区间负载代价：在RDD数据集转换和连接操作中，单个处理区间完成连接操作的代价，包括数据集合RD1和RD2从HDFS读取数据和分区传输、处理区间的连接计算以及连接结果存储，表达式 Costsum=Costio（RD1+RD2）+Costnetwork[Join（RD1+RD2）+Noise（RD1+RD2）]+CostJoin[Join（RD1）×Join（RD2）]。其中，数据量定义为采样得到的元组数量，代价定义为处理的数据量。Costio（RD1+RD2）表示RDD数据读取及分区shuffle的代价。Costneteork表示网络传输的代价，Join（RD1+RD2）表示处理区间中实际进行连接的数据量，Noise（RD1+RD2）表示处理区间中与连接无关的数据量 （连接噪声）。CostJoin[Join（RD1）×Join（RD2）]表示进行连接计算以及结果数据输出到HDFS的代价。Ri（Costsum）表示处理区间Ri对应的连接代价。数据倾斜问题的解决取决于各个处理区间负载代价的均衡程度，即满足R1（Costsum）≈R2（Costsum）≈…≈Rk（Costsum）。其中，各个处理区间关于Costnetwork和Costio的代价，与分配给处理区间的实际数据量是成正比的。当分配到各个处理区间的数据量达到均衡时，处理区间Ri之间的Costio和Costnetwork是近似相等的，这时处理区间的负载均衡条件转化为R （CostJoin）≈R2（CostJoin）≈…≈Rk（CostJoin），而Ri（CostJoin）取决于连接数据Join（RD1）×Join（RD2），因此，为了实现各个处理区间的负载均衡，除了实现数据量的均衡分配，同时也要保证连接数据或连接噪声的均衡。

3　虚拟区间划分算法VPRP

3.1算法描述

本文在充分考虑连接噪声对负载均衡影响的基础上，根据区间划分的思想[6]，提出了基于虚拟处理区间划分和交叉映射的连接算法，基本思想如下：首先通过文献[4]提出的采样方法，分别获取数据集RD1和RD2在连接属性Key的数据分布情况，数据格式为{（K1，C1），（K2，C2），…，（Kn，Cn）}，通过对样本集（RD1）sample和（RD2）sample的Key值排序和投影比较，估计出两个样本集连接数据集RDL和连接噪声集RDV，根据虚拟分区数量v，分别将数据集RDL和RDV以键值数量Ci均匀分割v份，得到{L1，L2，…，Lv}和{V1，V2，…，Vv}，合并组装得到数据量和连接噪声负载均衡的虚拟区间H={L1V1，L2V2，…，LvVv}，采用轮转法建立虚拟区间H到处理区间R的交叉映射关系MapPartition{（H1，Hk+1，…，H（α-1）k+1→R1），（H2，Hk+2，…，H（α-1）k+2→R2），…，（Hk，H2k，…，Hαk→Rk）}，最后根据映射关系分配数据，并完成连接操作。其中，虚拟区间数量v是处理区间数量k的整数倍（实验设定v=2×k=40）。

3.2算法详细流程

虚拟处理区间划分算法共分以下三个阶段：

（1）采样阶段

输入：连接数据集RD1和RD2

输出：样本集（RD1）sample和（RD2）sample，输出数据表示为{（K1，C1），（K2，C2），…，（Kn，Cn）}，其中Ki表示连接属性上的键值，Ci表示与Ki对应的元组数量。

①Map任务读取数据的过程中，Master节点分别从RD1和RD2数据集对应的集群中，随机抽取50%比例的计算节点，为选中的节点开启独立的统计进程，从中读取20%的数据集Ti。

②各个统计进程对读入的Ti按照Ki值排序，并统计相同Ki值的数量Ci，得到中间数据 {（K1，C1），（K2，C2），…，（Ki，Ci）}，按照Ci从大到小逆向排序，并从中抽取前p个最大的（Ki，Ci）集合Slargest，然后从剩余的数据中随机抽取q个（Ki，Ci）集合Snormal，根据Ssample=Slargest∪Snor-mal组成节点的样本集。

③Master节点开启一个Reduce作业汇总各个统计进程采集到的样本集Ssample，按照Ki值升序排列，相同Ki值的Ci相加合并，最终得到输出样本集（RD1）sample和（RD2）sample。

（2）虚拟区间划分阶段

输入：样本集（RD1）sample和（RD2）sample

输出：虚拟区间H={L1V1，L1V1，…，LvVv}

①对样本集（RD1）sample和（RD2）sample的 Key值进行投影比较，使用和，估计出连接数据集RDL和连接噪声集RDV。为降低采样误差，造成个别共有Key值划入RDV中的概率，将RDV中低于阈值θ（θ参考值为4～9）的数据区间和单Key值所对应的（Ki，Ci）按序插入到RDL中。

②根据虚拟分区数量v，分别将数据集RDL和RDV以键值数量Ci均匀分割v份，得到连接数据分割区间{L1，L2，…，Lv}和连接噪声分割区间{V1，V2，…，Vv}。

分割算法：

输入：数据集RDL和RDV，用U={U[0]，U[1]，…，U [n-1]}表示

输出：区间{L1，L2，…，Lvv}和{V1，V2，…，Vv}，用P={P [0]，P[1]，…，P[v-1]}表示

1.Cur_key←U[0].K Cur_size←U[0].CP[0]←U [0]i←0 j←0

2.Avg_size←Sum（U[0].C+U[1].C+…+U[n-1]. C）/v

3.while i＜v do

4.while U[j+1].C/2＜=Avg_size-Cur_size do

5.Cur_size←Cur_size+U[j+1].C

6.P[i]←P[i]U.[j+1]

7.j++

8.end while

9.i++

10.end while

（3）映射关系建立和执行连接阶段

输入：分割区间{L1，L2，…，Lvv}和{V1，V2，…，Vv}

输出：映射 MapPartition{（H1，Hk+1，…，H（α-1）k+1→R1），（H2，Hk+2，…，H（α-2）k+2→R2），…，（Hk，H2k，…，Hαk→Rk）}

①合并区间{L1，L2，…，Lv}∩{V1，V2，…，Vv}，得到虚拟区间H={L1V1，L2V2，…，LvVv}

②采用轮转法建立虚拟区间H到处理区间R的交叉映射关系MapPartition{（H1，Hk+1，…，H（α-1）k+1→R1），（H2，Hk+2，…，H（α-1）k+2→R2），…，（Hk，H2k，…，Hαk→Rk）}，其中Hαk=Lv∪Vv。

③把映射关系MapPartition用分布式缓存机制发送到各个mapper节点上。

④每个mapper根据射关系MapPartition判断数据的划分，将其对应的划分编号关联到（Ki，Ci）对上，然后划分器会将其解码得到划分编号，最后根据划分编号把（Ki，Ci）对发送给对应的reducer。

⑤执行等值连接，得到结果。由于使用了相同映射关系，所以每个reducer上的RD1和RD1分区的K值是相同的，reduce任务作Hash连接，将RD1和RD1分区中Hash值相同的所对应的value进行连接。

4　实验及结果分析

实验在4台计算机虚拟出的Spark集群上进行，其中一个NameNode节点，一个SecondaryNameNode和20个worker节点，其中CPU为Intel四核i5-3470@ 3.2GHz，内存8GB，硬盘1TB 7500RPM，网络100Mbps。软件平台：操作系统Ubuntu 14.04 LTS 64，Spark 1.1.0使用yarn模式，Hadoop 2.4.1，Scala 2.10.4。文中使用Pavlo等人的基准测试程序[19]，来对比分析有倾斜数据的连接优化算法。具体查询语句如下：

SELECT sourceIP，AVG（pageRank），SUM（adRevenue）as totRevenue

FROM rankings AS RK JOIN uservisits AS UV ON RK.pageURL=UV.destURL

WHERE UV.visitDate BETWEEN date（xxx）AND date（xxx），GROUP BY UV.sourceIP

实验一，该基准测试用于计算每个IP在某段时间访问该网站的总广告收入，数据集包含rankings表10GB和uservisits表200GB，为分析不同倾斜数据对性能的影响，从给定的数据集中分别抽取四个不同时间段的数据，每段时间的数据倾斜度不同。每个时段执行时间为6次查询的平均数。

图1　不同倾斜度下查询时间对比图

如图1所示，横坐标表示分别取查询中的不同时间段，纵坐标表示查询优化前后的运行时间。由于时间段1与时间段4的区间数据倾斜不太严重，所以系统性能不太明显；而时间段2与时间段3的数据倾斜严重，所以性能提升显著。

实验二，为考虑连接噪声是否均衡对整体性能的影响，选取时段3的数据，分别按照比例15%、35%、55%、75%、95%随机采样，对各个样本集添加无用的噪声数据达到与源数据相同的规模，使得各个样本集的数据倾斜度近似相同，而连接噪音逐步递减。给出不考虑连接噪声影响的算法VPRP-T（即将虚拟区间划分阶段的输入改为样本集（RD1）sample（RD2）sample，且略过Step1）与VPRP进行对比实验。

如图2所示，横坐标表示参与连接数据占总数据量的比值，纵坐标表示查询优化前后的运行时间。数据倾斜度和数据规模不变的情况下，连接数据所占比例与查询时间成正比。连接数据所占比值太大或太小（低于15%或高于95%），其分配的均衡程度对整体性能的影响较小，反之，连接计算量分配的不均衡，对整体的性能影响相对较大。综上可知：当数据倾斜严重时，本文提出的基于虚拟处理区间划分算法VPRP，由于考虑了节点间的数据量和实际连接数据的均衡分配，性能提升明显。

图2　相同规模数据不同连接率查询时间对比图

5　结语

本文给出了一种基于虚拟处理区间划分和交叉映射的连接算法VPRP，该算法通过虚拟处理区间划分和交叉映射，保证了shuffle阶段数据量和连接噪声的均匀分配，实现了各个处理区间的负载均衡。该算法与Hash连接算法在Spark集群上进行对比实验，结果表明该算法能有效解决连接中的数据倾斜问题。但是该算法并没有考虑如何优化分配Reduce任务，提高数据本地化程度，这将成为下步研究的重点方向。

[1]Blanas S，Patel J M，Ercegovac V，et al.A Comparison of Join Algorithms for Log Processing in MapReduce[C].Proceedings of the 2010 ACM SIGMOD International Conference on Management of data.ACM，2010.

[2]卞昊穹，陈跃国，杜小勇，等.Spark上的等值等值连接优化[J].华东师范大学学报：自然科学版，2014.

[3]翟红敏，刘国华，赵威，等.MapReduce中等值连接负载均衡优化研究[J].计算机工程与科学，2014.

[4]吴磊.基于hadoop的等值连接算法中数据倾斜问题的研究[D].中国科学技术大学，2014.

[5]Chen Q，Yao J，Xiao Z.LIBRA:Lightweight Data Skew Mitigation in MapReduce[J].IEEE Transactions on Parallel&Distributed Systems，2015.

[6]Atta F，Viglas S D，Niazi S.SAND Join—A Skew Handling Join Algorithm for Google's MapReduce Framework[C].Multitopic Conference（INMIC），2011 IEEE 14th International.IEEE，2011.

Equi-Join;Load Balancing;Data Skew;Range Partition

A Load Balanced Equi-Join Algorithm Based on Virtual Processor Range Partition

HU Zhong-kui1，QU Bo2，HUANG Bin1，LI Wen-yang1
（1.College of Computer Science，Sichuan University，Chengdu 610065）2.78098 PLA Troops，Chengdu 611200）

1007-1423（2016）03-0003-05

10.3969/j.issn.1007-1423.2016.03.001

胡忠奎（1981-），男，辽宁海城人，硕士研究生，研究方向为大数据分析

屈波（1976-），男，成都崇州人，讲师，研究方向为分布式和高性能计算

黄斌（1986-），男，云南曲靖人，硕士研究生，研究方向为信息安全

黎文阳（1990-），男，河南信阳人，硕士研究生，研究方向为分布式与数据库

2015-12-18

2016-01-15

数据分析和处理是大数据处理中最重要的任务，而等值连接又是数据分析中最常用、代价最高的操作之一。在实际的等值连接操作中，存在一个重要的问题就是数据倾斜：分配到每个任务的数据量不均衡，造成部分任务的完成时间更长，致使连接性能受到严重影响。为解决这个问题，提出一种负载均衡的等值连接算法（VPRP），通过采样估计数据集在连接属性上的数据分布情况，并采用虚拟分区和交叉映射的方法，在倾斜严重的数据周围划分出更多的区间，以增加数据分配的均衡性，同时消减连接噪声对整体性能的消极影响，最后实验验证该算法的有效性。

等值连接；负载均衡；数据倾斜；范围分割

Data analysis and processing is one of the most important tasks in large-scale distributed data processing applications.Join operations is one of the most common and costly in data analysis.One significant issue in practical Join operations is data skew:the imbalance in the amount of data assigned to each task.This causes some tasks to take much longer to finish than others and can impact performance.To solve data skew,presents a load balancing Join algorithm.It uses sampling method which can achieve a highly accurate approximation to the data distribution of connection properties.It can detect the data which is useless for the join operations to reduce Negative influence so as to realize load balancing.It uses the method of the virtual processor range partition and cross mapping to control the data skew which can improve the efficiency of equl-Join obviously.The simulation and experimental results show the algorithm proposed efficient.