大规模数据密集型系统中的去重查询优化

大规模数据密集型系统中的去重查询优化宋怀明1,2 安明远1,2 王洋1,2 袁春阳3 孙凝晖11(中国科学院计算技术研究所计算机系统结构重点实验室, 北京 100190)2(中国科学院研究生院, 北京 100049)3(国家计算机网络应急技术处理协调中心, 北京 100029)(shm@)Duplication Elimination in Large Scale Data Intensive SystemSong Huaiming1,2 An Mingyuan1,2 Wang Yang1,2 Yuan Chunyang3 Sun Ninghui11(Key Laboratory of Computer System and Architecture, Institute of Computing Technology, Chinese Academy of Sciences, Beijing 100190)2(Graduate University of Chinese Academy of Sciences, Beijing 100049)3(National Computer Network Emergency Response Technical Team/Coordination Center of China, Beijing 100029)Abstract As the emerging data intensive applications have received more and more attentions from researchers, it’s a severe challenge for duplication elimination for large volume data in a cluster environment. This paper proposes an effective and adaptive data placement method which is a combining of hash partitioning and histogram, as well as a design of an asynchronous parallel query engine (APQE) for duplication elimination. The APQE method develops maximum degree of pipeline parallelism for large scale data processing, and eliminates synchronous overhead of multiple nodes parallel, and returns part of query result as early as all data dependency eliminated. Experimental results demonstrate that the combined data placement method works well for relative attributes duplication elimination, and the asynchronous parallel query engine can make a great performance improvement in duplication elimination of large volume of data in a cluster environment.Keywords duplication elimination; data partitioning; large scale data intensive system; asynchronous query; parallel query engine(PQE)摘 要 针对shared-nothing结构下大规模数据密集型系统去重查询的挑战,提出了一种有效的数据分布策略和并行处理方法分别对相关属性和无关属性去重进行优化:即自适应的散列和直方图相结合的数据分布策略,以及异步式并行查询中间件.前者在数据写入时保证数据均衡,并在数据量发生倾斜时自动调整数据的分布;后者充分发掘了去重查询处理中的粗粒度流水级并行,并消除了多节点同步等待的开销,尽早地返回结果.在生产系统DBroker上的测试表明,数据分布策略极大地改善相关属性的去重查询性能,而异步式并行查询引擎能够充分发掘并行性,对不相关属性的去重查询具有显著的性能提升.关键词去重查询；数据划分；大规模数据密集型系统；异步查询；并行查询引擎中图法分类号 TP3921．引 言近年来,大规模数据密集型系统越来越多的应用在了网络安全监控、金融数据分析、电信数据处理、传感器网络等领域.在大规模数据密集型系统中,底层通常采用shared-nothing的结构将数据分散存储在多个节点,项目基金:国家“八六三”高技术研究发展计划基金项目（2007AA010505）上层采用并行查询中间件的方式,这种方式易于构架和实现,具有较好的并行性和扩展性,但另一方面数据的分散存储也给全局的数据去重查询提出了新的挑战,主要表现在以下几个方面:z节点间大量数据交换对网络通讯的压力.z大量的中间结果的存储和管理.z多节点并行带来的执行倾斜和同步等待的问题.在基于shared-nothing结构的系统中,数据通过划分,分布在多个节点[1,2],在去重查询中,将划分属性和去重属性相同的查询叫做相关属性去重,否则叫做不相关属性去重.Hash分区方法对于相关属性去重具有较好的效果,但在系统的数据集动态变化时,节点间数据量的动态均衡成为一个突出问题;在不相关属性去重查询处理中,常采用hash重分区的策略,但容易造成节点之间大量的数据交换,给网络通讯带来巨大压力.在基于中间件实现的大规模数据查询处理中,目前比较成熟的是两阶段查询处理方法[3,4],即前处理和后处理两个阶段:前处理在各数据库节点独立的运行,并将中间结果进行物化存储,后处理对前处理的物化存储的结果进行合并.但在中间结果数据量较大时,物化的方式严重影响了查询性能.此外,多节点并行不可避免出现执行倾斜,同步等待的方式可能导致大量节点资源的空闲等待.针对大规模数据去重查询的挑战,提出了一种自适应的散列和直方图相结合的数据分布策略,动态的调整各节点之间的数据均衡,适合于数据集动态变化的系统,如流数据的离线分析系统.同时也设计实现了针对海量数据处理的并行查询中间件,充分发掘并利用机群环境下的多机流水并行,提高了查询处理的运行效率,并通过异步方式的改进,消除多个并行节点之间的同步等待,提高流水处理的效率,在真实系统DBroker上的测试表明,这些方法均达到了较好的效果.本文的组织结构如下:第2节对介绍了数据划分和去重查询的相关研究;第3节提出了针对shared-nothing 结构的海量数据的去重提出了一系列的优化方法,包括自适应的hash和直方图相结合的数据划分策略,并行查询引擎的设计以及异步方式的改进;第4节在生产系统DBroker进行了相关的实验,证明划分策略以及并行查询引擎在海量数据去重查询处理中的有效性;第5节是对全文的总结.2．相关研究去重(duplication elimination)是数据库中的常见查询操作,目前比较成熟的消除重复的算法主要有排序合并算法[5,6,7]和散列合并算法[7,8],而在机群结构的大规模数据密集型系统中,网络传输和通信开销增加了去重查询处理的难度.为了减少去重查询中节点间的通讯,常采用hash方法对数据进行划分，需要解决数据均衡问题,为了避免数据倾斜问题,文献[9]提出了一种桶扩展(bucket spreading)的hash划分,文献[10]提出了一种基于等深直方图的hash划分.当节点间数据量发生倾斜时,文献[11]提出了一种采用重划分的策略动态的调整数据平衡的方法,文献[12]则采用了一种部分重划分的方法解决倾斜问题.而本文采用的自适应的hash和histogram 相结合的划分策略,借鉴了以往的数据划分思想,尽量避免节点间的数据倾斜.并在数据倾斜时自适应地进行调整,增加了全局的统计信息,不需要对数据进行重新计算和划分,减少了节点间的数据传输量.数据划分能够有效减少相关属性去重时节点间的数据传输,提高查询性能,但对无关属性的去重是无效的,而数据重划分的方法[13,14]则会引入大量的数据传输.文献[3,4]均提出了一种通用的两阶段的查询处理方法,通过物化中间结果的方式,能支持大部分查询处理,但对于中间结果巨大的去重查询来说,物化方式效果并不理想.采用shared-nothing结构的数据库系统,无论采用哪种数据分布,也无论数据量多么均衡,都不可避免各节点执行时间不一致的情况.其主要原因可归于三点:(1)节点性能的差异.由于硬件缺陷或者不稳定导致部分节点的执行性能低下,在机群环境下,不可避免的会出现个别节点的硬件发生老化或者缺陷的情况.(2)任务自身的差异.相同任务在不同的节点可能产生不同数量的中间结果,带来计算量和存取开销的差异.文献[13,15]均描述了由于各阶段任务划分差异导致执行时间不一致的现象.针对这些中间结果倾斜的情况,文献[13,14]均提出了任务重划分的方法,但是在大规模数据密集型系统中,任务的重划分会引入大量的网络数据的交换,在中间结果巨大的情况下,这种方法并不合适.(3)任务调度的差异.多任务同时运行时,由于各节点调度顺序的差异,导致每个任务在各节点的完成时间的差异,王[3]对这种由调度顺序的差异带来的执行倾斜展开了研究.文献[16,17]等提出了gang scheduling的调度方式,将一些需要同步进程或线程同时调度执行,能减少同步等待时间,但是如果调度的请求中含有I/O或通信等操作,调度效果仍不太理想,而海量数据处理中I/O开销更为突出,因此这种方法也不适用.多节点的执行时间存在差异是不可避免的,文献[4]提出了一种同步区的技术,在两阶段查询处理过程中,后处理需要等到所有数据库节点返回结果再进行结果的合并,同步区的技术放弃某些执行时间过长的节点,以提高整个查询的响应速度.文献[18]提出了一种异步流水化的处理方式,将部分中间结果实体化并共享这些数据,使得其他处理器也能对其操作,避免空闲等待.而本文提出的异步式的并行查询引擎是基于中间件的实现,采用了粗粒度的流水处理方式,没有同步区的概念,保存并共享所有计算的中间结果,各节点数据返回时自动触发计算,各节点的执行和数据返回是相互独立的,避免了同步等待开销.同时异步的执行方式能够尽早的返回用户的部分查询结果,大大减少了用户请求的响应时间,同时释放相应的存储空间,在海量数据处理过程中,极大的提高了查询执行的效率.3．系统设计和实现3.1 自适应的hash和histogram结合的数据划分消除重复常用的方法有排序方法和散列方法,散列方法的基本思想是将数据分成多个散列桶(bucket),然后在各桶内用同样的方法消除重复,这是一个递归的过程.如果已知用户查询的常用去重列,在数据写入时对数据进行散列划分,则可节省查询时划分的开销.Hash和histogram相结合的基本思想是将数据首先进行hash运算,再按照等深直方图对hash结果进行划分,得到均衡的数据分布.常见的hash函数是取模的运算,假设节点数为n,取一个较大的数m(m>>n),将记录的R的属性值col a与m取模Mod (R.col a, m), 可能得到 0~m-1, 这m种可能结果,然后对各hash值的记录数进行统计,再按照等深直方图的方法分成n份,结果如下:S={S1(0,x1), S2(x1,x2), …, S n(x n-1,m-1)}其中S i(i=1～n)表示分成的散列集合,每个集合对应一个数据库节点,存放对应的记录.其中m值远大于节点数的原因在于细化hash的结果,避免生成偏斜的直方图.结合对历史数据的统计,从hash结果得到直方图的方法有很多[9,10],这里不再展开.Hash和histogram结合的数据划分,建立在一定的统计基础上,能够保证各节点间数据写入的均衡,但是随着各节点数据库运行状态的变化(删除操作,或写入数据的分布变化),难免出现倾斜的情况,此时需要修改分布策略,进行部分的数据迁移达到新的均衡.自适应的改进是对数据倾斜的调整策略,增加了全局的hash分组计数统计,数据调整包括以下3个步骤:(1)利用统计信息计算各节点上数据量,并将偏差超过平均数一定比例的节点分成两组,正的记为{N1,N2,… N p},负的记为{M1,M2,… M t};(2)将每个超过节点N i的hash分组计数按照值大小排序得N i1,N i2,N i3…,逐个计算临界hash分组N ik,使得∑=≤kjAijSN1and∑+=>11kjAijSN成立,S A表示各节点平均数据量,该节点保留N ik之前的分组;(3)采用装箱算法将所有超出节点余下的hash分组分配到节点集合{M1,M2,… M t},同时修改全局的hash分组信息表.自适应的方法不需要重新进行hash划分,也不需要进行分组计数,只需要利用全局的统计信息,同时使得节点间的数据量传输总量达到最小.3.2并行查询引擎及其异步方式的优化 Hash和histogram相结合的数据划分,避免了节点间的数据传输,提高了相关属性的去重查询的性能,但是对于非相关属性的去重查询,并不能带来真正的好处.本文引入了并行查询引擎的设计,对传统两阶段查询进行优化,取消了物化中间结果的步骤,充分发掘多节点之间的粗粒度的流水级并行,对不相关属性的去重查询具有明显的性能提升,其查询处理结构如图1所示.Fig.1 Parallel Query Engine Architecture图1:并行查询引擎结构图并行查询引擎是大规模数据库的查询中间件,为每个查询分配一个master对象和一组slave对象,其中master对查询处理进行控制,完成SQL的分析和下发,启动结果合并线程,并返回最终结果;slave线程则在各数据库节点运行前处理查询语句,并将结果传送给master.并行查询引擎具有良好的流水并行性,各slave返回结果和master的结果合并,以及最终结果的输出可以同时进行,避免去重查询中大量的中间结果的物化过程,流水的处理方式大大降低了处理过程中存储的需求.为了粗粒度的流水处理能够连续进行,需要对去重查询的SQL语句进行改造.比如在在网络安全监控系统中,对各出入口的源地址的去重数量统计select count(distinct src_ip), inout_id from event_base where … group by inout_id;须在前处理语句增加对出入口属性的排序,即前处理语句改写成select src_ip, inout_id from event_base where … group by inout_id,src_ip order by inout_id.增加对分组属性的排序,在各节点的查询中,数据库内部的算法从hash group变成了sort group,算法效率上并没有太大的差异,但这种改变却使得后处理的结果合并能够流水的进行,并同时输出结果,提高响应速度,此外master不需要保存大量的中间结果,降低了运行时的存储消耗.在shared-nothing的大规模数据库系统中,各节点执行时间的差异不可避免,而同步等待的方式会导致大量的系统资源的浪费.异步的方式正是针对各节点执行时间差异进行的改进,防止流水线在启动或者运行过程中过多的同步等待.主要体现在两点:(1)Master不必等到所有的节点前处理都运行完成才启动合并线程;(2)在流水的对分组结果合并时,也不必等到所有节点都返回相关分组数据才进行合并计算.只要有部分节点返回数据就可以启动合并计算.异步的方式对slave返回和master合并结果进行了改进,其算法如表1所示.Table 1 Algorithm of Asynchronous Parallel Query Engine表1 异步并行查询引擎的算法简要描述Slave线程读取结果并分发Fetch_deliver(){…while ( !rs->eod()) { // 未到结尾hash_map.add (rs.record()); // 结果放入hash队列cur_top (); // 修改当前结果标志rs->next(); // 取下一条结果}}Master线程合并最终结果Merge(){…while (! hash_map.empty() ||! rsv->eod()){ // 当hash_map不为空或结果未取完cur = get_top(); // 获取当前的结果集顶端count_dinstinct(cur);//计算所有的小于顶端的分组// 的去重统计结果}}异步查询引擎采用多维的hash map保存中间结果,每个slave独立运行并将返回的结果放入hash map,不必关心其他slave的状态.Slave将中间结果写入hash map 时自动消除重复,因此计算是由各slave的数据返回触发的.此外各slave返回的数据是有序的,因此对于每个分组,只要所有slave都完成了该分组数据的返回, master就可以输出该分组的合并结果,释放相应的空间.在处理过程中,master的结果输出和slave的数据返回是相互独立、并行运行的.异步的方式是对读取结果和合并计算的优化,消除了节点间的同步等待,提高系统资源的利用率和查询处理的效率.在实现中,为了避免计算中的对共享结构hash map的锁冲突,对hash map进行了划分,同时采用批量写入的策略,减少了对单个资源的锁的次数.4．实验结果4.1实验环境介绍实验是在DBroker(863-917网络安全监测平台)系统上进行的,该系统是构架在Dawning 4000L机群服务器上的大规模数据密集型的应用系统.DBroker底层的数据库采用的shared-nothing的方式,其节点配置为(4*2.2G CPU,32GB Memory,4TB Raid5,1Gb Ethernet, Linux AS4, Oracle 10.2),查询服务器的节点配置为(4*2.2G CPU, 32GB Memory,10Gb Ethernet,Linux AS4).实验中采用了16个数据库节点和1个查询服务器节点,查询引擎运行在查询服务器节点.测试的数据集采用该系统的仿真数据,语句形式为select count(distinct src_ip), count(distinct dst_ip), inout_id, dev_id from event_base where … group by inout_id, dev_id; 表示对源IP和目的IP地址按照出入口和设备号进行去重计数统计.性能测试包括节点数扩展测试和数据量扩展测试,在节点数扩展测试中,每个节点的数据量是1200万条记录,节点数从2增加到16,每次增加2个节点;在数据量扩展测试中,保持节点数16不变,每个节点的数据量从400万增加到3200万条记录,每次增加400万条.4.2性能对比图2是对几种数据分布策略执行时间的比较,其中round-robin为轮询的数据分布,hash-1指不相关属性的去重查询,hash-2指相关属性的去重,两者均为散列和直方图相结合的划分方式.测试的节点数为16,每个节点数据量为从400万增加到3200万条记录,从测试结果可以看出,hash分布和等深直方图相结合的方式,能够大大降低相关属性的去重的响应时间,而对于不相关属性的去重和轮询方式的差别不明显.Fig.2 Execution time in different partition methods图2:不同分布策略执行时间的对比Fig.3 Execution time in no balance and rebalance图3:无平衡和自动平衡情况下的执行时间图3是对在不同数据倾斜程度下,采用自适应平衡策略后的执行时间和无平衡情况下的执行时间对比.采用了16个节点,每个节点平均1200万条记录,其中数据倾斜度表示最大数据量节点超出平均的数据量和总数据量的比值,即(max(ds)-avg(ds))/sum(ds),其中ds表示各节点的数据量.由测试结果可以看出,数据的倾斜程度越严重,在无平衡的情况下,执行时间越长,并且执行时间呈现超线性增长;而采用了自适应的方式,不管数据是多么不均衡,调整后系统的执行时间基本保持不变,自适应的数据分布方式能主动调整数据的均衡.Fig.4 Total execution time in different number of DB nodes图4:不同节点数下的执行总时间对比Fig.5 Execution time in each step图5:不同节点数下各阶段的执行时间对比图4和图5是对并行查询引擎和两阶段查询方法的节点数扩展测试.图中pqe表示并行查询引擎的方法, pqe-fore和pqe-post分别表示前后处理的时间;2step表示两阶段的处理方式,2step-fore和2step-post分别表示前后处理的时间.从测试结果可以看出,无论从总的执行时间,还是分阶段的时间来看,并行查询引擎的性能都比两阶段的执行方法更好.在并行查询引擎中,首条结果在所有节点前处理完成即可返回,随着节点数的增加,并行查询引擎的前处理时间是相对固定的,即首条结果输出的时间相对稳定;而两阶段的方式其前处理完成后有一个物化中间结果的步骤,处理时间随着节点数增加而不断增加,因此并行查询引擎具有更好的节点数扩展性.Fig.6 Total execution time in different data size图6:不同数据量下的执行总时间对比Fig.7 Execution time in each step in different data size图7:不同数据量下各阶段的执行时间对比图6和图7是对数据量的扩展测试.从总执行时间来看并行查询引擎的性能无论在小数据量还是大数据量的情况下,均优于两阶段的查询方法.而从各阶段的执行时间来看,在数据量较小的时候,两种方式前处理的时间差异不大,这是因为查询中间结果较小,而两阶段法物化中间结果的开销较小,但在数据量较大的情况下,并行查询引擎的性能更好,这是因为并行查询引擎省略了中间结果的物化过程,而是在合并计算中流水的读取各节点的前处理结果并进行去重.Fig.8 Total execution time in different number of DB nodes图8:不同节点数下各方法执行时间对比Fig.9 Total execution time in different data size图9:不同数据量下各方法执行时间对比图8和图9测试对并行查询引擎采用异步方式的改进后对查询性能的影响,图中用asyn表示.从测试结果可以看出,无论在增加节点数,还是增加各节点数据量的情况下,异步的方式都进一步提高了查询的性能.特别是在节点数逐步增加的情况下,相比同步的方式,查询性能提升效果更加显著.这是因为随着节点数的增加,各节点之间的同步等待开销所占的比例增大,而异步的方式的合并计算是由各节点的结果返回主动触发的,避免了节点间相互等待的开销,因而具有更好的扩展性. 4.3 讨论Hash和histogram相结合的数据划分,在保证各节点数据量均衡的情况下,使得相同属性值的数据存储在同一个节点,在相关属性的去重查询中,各节点去重的结果即是最终结果的一部分,不必进行合并计算,对于性能的提升是明显的.自适应的机制在数据量发生倾斜的情况下,不用额外增加划分和统计,对少量数据进行迁移,使得节点间数据量达到新的均衡.并行查询引擎是机群环境下大规模数据库系统的查询中间件,将两阶段查询处理中独立的先后步骤改造成多机协同的流水处理过程,并充分挖掘流水线各步骤间的并行性,对中间结果巨大的无关属性的去重查询具有较大性能提升.异步方式减小了节点之间同步等待的开销,进一步提高了查询结果合并的速度,提高系统资源的利用率,和同步的方法相比,具有更好的性能和扩展性.异步方式也存在一定的不足,由于需要保存更多的中间结果,内存的消耗比同步的方式要多一些.基于中间件的实现,前处理能较好的利用成熟的商业数据库的查询处理机制,不必考虑其内部具体的算法,简化了设计和实现.但基于数据库的数据存取和查询处理也存在一定不足,无法在更细粒度的层面参与处理过程,限制了某些具体的优化方法的展开.5．结论和进一步研究海量数据的去重查询已成为大规模数据密集型系统的一个关键应用.Hash和histogram相结合的数据分布策略,在减小节点间通讯量的同时,也能保证节点间的数据量均衡,大大提高了相关属性去重查询的性能,而自适应的方式在节点间发生数据倾斜时,能够进行自动的调整.并行查询引擎的设计通过流水处理的方式提高查询处理性能,异步方式在此基础上进一步消除了节点间同步等待的开销,对于非相关属性的去重查询,具有较好的效果.实验表明,这些方法能够大幅提高去重查询的性能.此外,这种数据分布策略和异步式的并行查询引擎同样适用于shared-nothing结构上的其它类型的查询,如分组查询(group)和连接查询(join).并行查询引擎能有效的进行去重查询,但目前还暂不支持去重后再进行全局操作.在下一步的工作中,将对去重后的全局计算展开研究,此外还需对自适应划分策略重新平衡数据量的代价进行细致的分析.参 考 文 献[1]Manish Mehta, David DeWitt. Data placement inshared-nothing parallel database systems. The VLDBJournal(1997)6:53-72, 1997.[2]David DeWitt and Jim Gray. Parallel database systems: thefuture of high performance database systems. Communicationsof ACM, V ol.35, Issue 6, pages: 85-98, 1992.[3]Wang Yong. Workload and Query Scheduling of Event StreamAnalysis: [PhD thesis]. Beijing, Institute of ComputingTechnology, Chinese Academy of Sciences, 2008.(in Chinese)（王勇. 事件流应用的负载特征分析和查询调度[博士论文]. 北京,中科院计算所,2008）[4]Jiao Limei, On Design Research of Event-Stream OrientedCluster Database[Ph.D Thesis], Beijing, Institute ofComputing Technology, Chinese Academy of Sciences,2007.(in Chinese) (焦丽梅. 面向事件流机群数据库系统的设计研究[博士论文]. 北京,中科院计算所, 2007)[5]Dina Bitton, David J. Dewitt. Duplication Record Eliminationin Large Data Files. ACM Transaction on Database Systems,V ol.8, No.2, pages: 255-265, June 1983.[6]Xiaoyu Wang, Mitch Cherniack. Avoid Sorting and GroupingIn Processing Queries. In: Proc of the 29 VLDB conference.Berlin, Germany, 2003.[7]J. Claussen, A. Kemper, D. Kossmann, C. Wiesner. Exploitingearly sorting and early partitioning for decision support queryprocessing. The VLDB Journal(2000)9: 190-213, 2000.[8]Goetz Graefe, Richard L. Cole. Fast Algorithms for UniversalQuantification in Large Databases. ACM Transactions onDatabase Systems, V ol.20 No.2, pages:187-236, June, 1995. [9]Masaru Kitsuregawa, Yasushi Ogawa, Bucket SpreadingParallel Hash: A New, Robust, Parallel Hash Join Method forData Skew in the Super Database Computer(SDC), VLDB90,pages:210-221, 1990.[10]Ung Kyu Park, Hwang Kyu Choi, Tag Gon Kim, Uniformpartitioning of relations using histogram equalizationframework: an efficient parallel hash-based join, InformationProcessing Letters, V ol. 55, No.5, pages:283-289,1995.[11]Kien A. Hua, Chiang Lee, Handling data skew inmultiprocessor database computers using partition tuning,VLDB91, pages:525~535, 1991.[12]Yu Xu, Pekka Kostamma, etc. Handling data skew in paralleljoins in shared-nothing systems. SIGMOD08, pages:1043~1052, 2008.[13]Lionel Brunie, Harald Kosch. Control strategies for complexelational query processing in shared nothing systems.SIGMOD Record, V ol.25, No.3, September 1996.[14]Ambuj Shatdal and Jeffrey F. Naughton. Adaptive parallelaggregation algorithms. ACM SIGMOD Record 1995, V ol.24,Issue 2, pages: 104-114, May 1995.。