基于MapReduce的Skyline-join查询算法

孙大烈，李建中

(哈尔滨工业大学 计算机科学与技术学院，150001哈尔滨，sdl@hit.edu.cn)

给定一个感兴趣的属性集合，Skyline查询返回这样的元组，这些元组在任何一个属性上都不受其他元组制约[1].例如，一名学生想要寻找一个合适的住处，他可能提交这样的查询:“返回价格便宜并且离学校距离近的公寓”.Skyline查询在决策系统中很有价值(用处).由于它的重要性，研究人员已经开始在商用数据库管理系统(DBMS)中实现 Skyline 查询[2－3].

现有的研究工作大多假设Skyline查询只局限于一个数据表，也就是说，所有待查的属性都来自于同一张数据表.然而，这种假设在互联网环境中不再成立，因为此时查询处理需要来自多源的数据.例如，数据库cheapoair.com提供机票预订服务，BookInHotels.com提供酒店预订服务.假设用户提交这样的查询“请列出所有在5月11日起飞的最廉价的航班，以及距离机场最近的四星级酒店”.这种查询与Skyline查询有共同的特点，但是它需要从多张数据表提取数据信息.在本文中，称这种类型的查询为Skyline-join.

处理Skyline-join查询的一个简单而原始的方法是，先连接所有相关的数据表，然后再应用现有的Skyline算法.然而，这种简单的算法往往效率低下，不能提供及时的结果.因此，本文提出一种新的基于 MapReduce框架[4－5]的分布式并行算法.该算法可以应用在计算机集群环境中，通过将计算分布在不同的节点上来提高处理速度.

1 Skyline-join查询

如前所述，不同于现有的操作，Skyline-join操作会涉及多张数据表.对于某单一数据表，如果当0≤i≤d时，vi≻v'i并且 ∃j(0≤ j≤d)∧(vj≻v'j)，这里定义≿和≻为偏序关系，则元组ti=(v0，v1，…，vd)制 约 元 组 tj=(v'0，v'1，…，v'd).Skyline查询返回那些不受其他元组制约的元组.若Q= ＜ {A1，A2，B1，B2}，Ø，{A3，B3}＞ 取自多张数据表，则引发了新问题——多数据表Skyline查询.为处理这种查询，本文引入一种新的查询操作，称为 Skyline-join.最近，Skyline-join查询引起了其他研究者的兴趣[6].

设有数据库D，其中数据表集合为T，Skylinejoin表示为 ＜S，C，J＞，其中S为Skyline查询中的属性集合，C为用户提交的查询条件集合，J为连接属性集合.S(T)，C(T)和J(T)分别为对于某数据表T的Skyline查询属性集、条件属性集以及连接属性集.为讨论简便起见，这里假设J∩S=Ø.事实上，J∩S≠Ø只是算法的一种特殊情况.对于某查询 ＜S，C，J＞，Skyline-join检索空间定义为:

定义1 Skyline-join检索空间.

Ttotal是查询 ＜S，C，J＞ 的Skyline-join检索空间，当且仅当

c)不存在T'total满足上述两条性质，且其中数据表的个数少于Ttotal.

Skyline-join查询在其相应检索空间里进行处理，检索空间中不同元组之间的制约关系定义为:

定义2 Skyline-join Domination.

给定Skyline-join查询 ＜S，C，J＞及其检索空间 Ttotal，对于 Ttotal中的两元组 t1=(v0，v1，v2，…，vd)和 t2=(v0，v1，v2，…，vd)，t1制约 t2当且仅当

为表述方便，本文用ti≻domtj表示元组tiSkyline-join制约元组tj，并用ti≻Atj来表示元组ti在属性集合A上Skyline-join制约元组tj.

定义3 Skyline-join结果集.

对于Skyline-join查询Q=＜S，C，J＞，检索空间为Ttotal，其结果集R有以下性质:

性质1 ∀ti(ti∈R)→¬∃tj(tj∈Ttotal)∧(tj≻domtj).

性质2 对于元组ti∈R，ti=(v0，v1，…，vd)，若存在atti∈C，则vi必须满足相应查询属性条件.

为方便讨论简单，假设查询条件C=Ø并只考虑数值型属性，域值范围[0，100]，并用“MAX”作为Skyline条件.

2 MapReduce框架

MapReduce是由Google提出的处理海量数据的并行处理平台.它可以应用在计算机集群之上，通过将数据和计算分布在不同的节点来取得高性能.MapReduce也是一个灵活的编程框架，它提供了两个接口函数:Map和Reduce.用户可以实现自己的Map和Reduce函数来完成相关的处理任务.

Map和Reduce函数的输入必须是一对(key，value).其中key用来表示数据的键值，而value则代表实际的数据.Map函数可以抽象为

对每一个输入的(key，value)对，Map函数进行相应的处理，并输出一串新的(key，value)对.这些新的(key，value)对被传送到不同的Reduce函数进行进一步的处理.而Reduce函数可以抽象为

Reduce函数收到来自不同Map函数的(key，value)对.在进行处理之前，它先对这些(key，value)对按照它们的key值进行排序，具有相同key值的对被集成为(key，list(v))，即一个key值对应一个value数组.(key，list(v))作为输入被传入Reduce函数，然后按照用户的处理逻辑生成结果.通常结果是一串值 (表示为list(v))，并被写为分布式文件系统，如 GFS[7]和 BigTable[8].

图1列出了基于MapReduce的统计词频的伪代码.在Map函数中，每次对一行的文本进行分析，将单词切分出来.对每一个单词w产生一个(key，value)对，即 (w，1)，表示该单词已经出现了一次.在Reduce函数，对同一个函数的统计被集成在一个(key，list(value))中.因此，只要遍历该数组，就可以知道这个单词出现的次数.

图1 MapReduce词频统计算法

3 基于MapReduce的查询方法

本文用表1，表2作为例子来阐述相关的算法.

表1 表R

表2 表S

其中表R和表S使用ID属性进行连接，而对于其他属性 (A1…Am以及 B1…Bn)，Skyline-join查询使用Max作为条件，要求返回在这些属性上不被其他记录支配的记录.

使用MapReduce来处理Skyline-join查询可分为:1)一个MapReduce任务被用来产生表的连接结果;2)连接的结果被用来作为另一个MapReduce任务的输入，而该任务产生多个表的Skyline-join结果.

3.1 产生连接结果

在第1个MapReduce任务中，Map函数从分布式文件系统读取两个表的数据，对于每一个记录，Map函数产生一个(key，value)对，其中该记录的ID值被用作为key，而其他值被作为value.这样一来，能够在ID上连接的表R和表S的记录将被发送到同一个Reduce函数.该Reduce函数将含有相同ID的记录集合在一起，采用基于内存的连接算法产生相应的表连接结果.在表连接结果产生之后，Reduce函数再调用基于内存的Skyline算法，比如BNL算法.被其他记录支配的记录被舍去，因为它们不可能成为最终的Skyline-join结果.其他记录被写入分布式文件系统.图2给出了相关伪码.

图2 MapReduce的Skyline-join算法

图2是针对两个表的Skyline-join.对于更多表的Skyline-join查询，该算法依旧适用.唯一的区别就是当处理有n个表的Skyline-join查询时，需要有n种Map函数，每种函数读取一个表的数据.同时在Reduce函数，需要对n个表进行连接操作.

3.2 产生Skyline-Join结果

第2个MapReduce任务读取上一个任务产生的结果，然后采用空间划分的方式来计算Skyline结果.假设Skyline-join查询要求在x和y属性上得到不能被支配的元组，图3给出了一种可能的空间划分方法.该方法将x和y平均划为4个区域，从而整个空间被划为4×4=16的子空间.如果一个记录位于子空间S(a，b)，那么它的x属性的值位于x的第a个区间，而它的y属性位于y的第b个区间.

图3 空间划分

在第2个MapReduce任务中，Map读取上一个任务的结果，然后为每一个记录判断其所在的子空间.假设下一个记录t在空间S(a，b)，那么Map为它生成的(key，value)对为(h(a，b)，t).其中h为一个哈希函数，给定两个值a和b，h(a，b)返回一个Reduce函数的ID.根据该ID，Map函数知道该(key，value)对应该传递给哪一个Reduce函数.采用这种方法，同一个子空间的记录都会发送给同一个Reduce函数.该Reduce函数调用基于内存的Skyline算法，可以计算得出该子空间的Skyline结果.

当第2个MapReduce任务完成后，分布式文件系统存储着每个子空间的Skyline-join结果.然而这并不是最终的结果.为了得到全局的Skyline-join结果，第3个MapReduce任务被提交，用来合并子空间的Skyline-join结果.在合并之前，本文采用一个预处理过程来删减掉不可能产生结果的子区间.

在图 3 中，假设子空间 S(3，2)，S(2，2)，S(3，3)和S(2，3)产生了一些Skyline-join结果.那么子空间S(1，1)可以被排除，因为即使它产生了一些结果，其结果也必然被上述4个子空间中的结果支配，从而不可能出现在最终结果中.相反，子空间S(3，1)，S(2，1)，S(1，2)和 S(1，3)不能被排除，因为它们的结果并不能被前面的子空间支配.

预处理之后，剩下的结果被Map函数读取，Map函数为所有的记录产生同一个key，使得它们都被发送到同一个Reduce函数.该 Reduce函数调用传统的Skyline算法，如文献[9－10]，来产生最终的结果.

4 实验结果

为了验证分布式算法的有效性，本文采用Amazon的EC2[11]平台搭建了一个集群环境，并根据文献[6]的描述生成了实验数据.实验数据包括两个数据表R和S.表R包含3个属性:员工编号、年龄和工资.表S包含3个属性:员工编号、经理编号和公司编号.所有属性均为整形，并符合均匀数据分布，其中表R和表S使用员工编号进行连接操作.在一个n节点构成的集群中，每个表的元组数为1 500 000 n.每个节点运行4个Map进程和2个Reduce进程.

图4展示了随着节点个数的增加，查询处理时间的变化.可以看到，本文的分布式算法基本不受节点个数以及数据量的影响(节点越多数据量越大).因此，该算法具有很好的可扩展性.如果要获得更好的查询性能或者能够处理更多的数据，只需要增加更多的节点即可.

图4 节点数目的影响

作为比较，图5显示了当节点增加数据量不变的情况下，查询的效率变化.可以看出，本文提出的Skyline-join算法非常适合应用在并发处理平台如MapReduce之上.当节点数量增加，查询的速度也随之变快，几乎达到线性递减的趋势.

图5 节点数目的影响

5 结论

1)通过将数据以及计算分布在不同的节点上，使用并行化的处理机制来提高性能.

2)在Map阶段采用分片剪枝的方法来降低复杂度.

3)通过在真实的云计算平台实验表明，该算法具有高可扩展性的特点.

[1]BORZSONYI S，STOCKER K，KOSSMANN D.The Skyline operator[C]//Proceedings of the 17th International Conference on Data Engineering.Washington，DC:IEEE Computer Society，2001:421－430.

[2]CHAUDHURI S，DALVI N，KAUSHIK R.Robust cardinality and cost estimation for skyline operator[C]//Proceedings of the 22nd International Conference on Data Engineering.Washington，DC:IEEE Computer Society，2006:64.

[3]EDER H，WEI Fang.Evaluation of skyline algorithms in PostgreSQL[C]//Proceedings of the 13th International Database Engineering＆Applications Symposium.New York，NY:ACM，2009:334－337.

[4]DEAN J，GHEMAWAT S.MapReduce:Simplified data processing on large clusters[J].Communication of ACM，2008，51(1):107－113.

[5]YANG Hung-Chih，DASDAN A，HSIAO Ruey-Lung，et al.Map-reduce-merge:simplified relational data processing on large clusters[C]//Proceedings of the 2007 ACM SIGMOD International Conference on Management of Data.New York，NY:ACM，2007:1029－1040.

[6]JIN W，ESTER M，HU Z J，et al.The multi-relational skyline operator[C]//Proceedings of the IEEE 23rd International Conference on Data Engineering.Washington，DC:IEEE，2007，1376 －1380.

[7]GHEMAWAT S，GOBIOFF H，LEUNG Shun-`Tak.The Google file system[J].ACM SIGOPS Operating Systems Review，2003，37(5):29－43.

[8]CHANG F，DEAN J，GHEMAWAT S，et al.Bigtable:A distributed storage system for structured data[C]//Proceedings of the 7th Usenix Symposium on Operating Systems Design and Implementation.Berkeley CA:USENIX Association，2006:205－218.

[9]BARTOLINI I，CIACCIA P，PATELLA M.Salsa:Computing the skyline without scanning the whole sky[C]//Proceedings of the 15th ACM International Conference on Information and Knowledge Management.Arlington，Virginia，2006:405 －414.

[10]CHAN C Y，ENG P K，TAN K L.Stratified computation of skylines with partially-ordered domains[C]//Proceedings of the 2005 ACM SIGMOD International Conference on Management of Data.New York，NY:ACM，2005:203－214.

[11]Amazon.Amazon Elastic Compute Cloud(Amazon EC2)[EB/OL].http://aws.amazon.com/ec2//192－5518875－2032964.