錢招明,王雷,余晟雋,宮學(xué)慶

(華東師范大學(xué)數(shù)據(jù)科學(xué)與工程研究院,上海200062)

分布“系統(tǒng)中Semi-Join算法的實現(xiàn)

錢招明,王雷,余晟雋,宮學(xué)慶

(華東師范大學(xué)數(shù)據(jù)科學(xué)與工程研究院,上海200062)

隨著新型分布式系統(tǒng)的使用范圍越來越廣,應(yīng)用不再滿足于僅使用主鍵訪問方式來讀取數(shù)據(jù),如何在這些系統(tǒng)中高效實現(xiàn)Join等復(fù)雜操作成為研究的熱點.本文介紹了如何基于Semi-Join算法在分布式系統(tǒng)中實現(xiàn)Join操作,提出了兩種獲取右表數(shù)據(jù)的方法,并通過實驗分析了該算法的性能.

分布式數(shù)據(jù)庫;Join操作;Semi-Join算法

0 引言

隨著云計算技術(shù)的快速發(fā)展,各種新型的分布式系統(tǒng)不斷涌現(xiàn),越來越多的應(yīng)用開始采用分布式架構(gòu)存儲和管理數(shù)據(jù).早期的NoSQL系統(tǒng)多數(shù)采用簡單的Key-Value模型存儲數(shù)據(jù),提供按鍵值(Get操作)或按鍵值范圍(Scan操作)訪問數(shù)據(jù)的方法.隨著分布式系統(tǒng)越來越廣泛的被應(yīng)用,應(yīng)用系統(tǒng)對數(shù)據(jù)訪問方法提出了更高的要求,如何在分布式系統(tǒng)中高效實現(xiàn)連接(Join)、聚合(Aggregation)等操作成為近期研究的熱點問題之一.

Join是傳統(tǒng)數(shù)據(jù)庫系統(tǒng)中的基本操作之一,在集中式環(huán)境下已經(jīng)有很多成熟的算法用于實現(xiàn)Join操作,例如,Nest Loop Join、Merge Join、Hash Join等.在分布式環(huán)境下,我們?nèi)匀豢梢允褂眠@些算法來實現(xiàn)Join操作,但受網(wǎng)絡(luò)傳輸延遲的影響,算法的執(zhí)行效率可能會非常差.Semi-Join是20世紀(jì)80年代提出的用于優(yōu)化分布式數(shù)據(jù)庫中Join操作的算法[1],本文將該算法應(yīng)用于新型的分布式系統(tǒng),用于優(yōu)化Join操作的實現(xiàn).在Semi-Join算法的實現(xiàn)流程中,本問題提出了兩種獲取右表數(shù)據(jù)的方法,并且通過實驗對算法的性能進行了分析.

論文的內(nèi)容安排如下:第1節(jié)介紹分布式系統(tǒng)中Join操作相關(guān)的研究工作現(xiàn)狀;第2節(jié)介紹了分布式系統(tǒng)的架構(gòu);第3節(jié)介紹了基于Semi-Join算法的優(yōu)化方案;第4節(jié)通過在不同規(guī)模的數(shù)據(jù)集和不同大小的結(jié)果集場景下的實驗驗證了本文實現(xiàn)算法的性能;第5節(jié)對本文進行了總結(jié).

1 相關(guān)工作

如何在分布式系統(tǒng)中實現(xiàn)Join操作是近年來受到廣泛關(guān)注的一個熱點問題[2-6].與傳統(tǒng)集中式數(shù)據(jù)庫系統(tǒng)中Join操作實現(xiàn)算法不同,在分布式系統(tǒng)中,影響算法性能的主要因素不再單單是磁盤IO,通訊開銷、數(shù)據(jù)混洗和并行程度等因素成為重要的考量指標(biāo).

目前,關(guān)于分布式系統(tǒng)中Join操作實現(xiàn)算法的研究工作主要可以分為兩大類.一類關(guān)注于將Join運算分解為多個任務(wù),利用Map/Reduce等計算模型進行并行計算.例如,文獻[7]研究如何將多個Join操作分解后在多個Map/Reduce任務(wù)中執(zhí)行;文獻[5]提出了一種在執(zhí)行Join操作時根據(jù)節(jié)點負(fù)載重新劃分任務(wù)的自適應(yīng)算法.另一類關(guān)注于查詢樹的執(zhí)行策略優(yōu)化,特別是針對多個Join操作的執(zhí)行優(yōu)化.例如,文獻[8]分析了使用Left-Deep、Right-Deep和Bushy Query Tree等不同策略的優(yōu)缺點;文獻[9]分析對比了多個worst-case最優(yōu)算法的性能邊界.本文在參考已有研究工作成果的基礎(chǔ)上,描述了一類簡單有效的分布式系統(tǒng)中Join操作的實現(xiàn)算法.

2 分布式系統(tǒng)中的Join運算

2.1 分布式系統(tǒng)

圖1.1所示為常見的分布式系統(tǒng)查詢架構(gòu),Client與Query engine,Query engine與Storage之間通過網(wǎng)絡(luò)互連,一般情況下Client與Query engine之間是一對一關(guān)系,而Query engine作為Client請求的處理節(jié)點,負(fù)責(zé)與底層的眾多分布式存儲節(jié)點(Storage)進行數(shù)據(jù)交互.

其中Query engine中重要功能組件如圖1.2所示,其工作流程為:Client的SQL請求經(jīng)過SQL Parser解析后生成SQL Logical Plan并通過Query Optimizer優(yōu)化后產(chǎn)生最終的SQL Physical Plan交由Execution執(zhí)行,而Execution負(fù)責(zé)與Storage進行數(shù)據(jù)的交互以及最終結(jié)果集的運算,如連接運算.

圖1 分布式系統(tǒng)查詢架構(gòu)與查詢引擎Fig.1Query architecture and query engine of distributed system

Storage一般提供兩種數(shù)據(jù)訪問方式:讀操作或?qū)懖僮?通常,Storage可提供讀操作也可提供寫操作,對于讀寫分離的分布式系統(tǒng)而言,Storage被分為讀節(jié)點與寫節(jié)點兩種角色.而特別針對讀操作,并且在通過主鍵訪問數(shù)據(jù)的場景下,本文考慮兩種不同的數(shù)據(jù)過濾方法:一是通過主鍵定位數(shù)據(jù)的Get方法;二是經(jīng)由主鍵范圍掃描數(shù)據(jù)的Scan方法.

2.2 分布式Join操作

圖2.1為分布式Join操作的執(zhí)行計劃,其最終在Query engine的Execution組件中執(zhí)行. Join算法一般分為Merge-Join、Nested-loop-Join和Hash-Join三類,本文以Merge-Join為例分析其數(shù)據(jù)請求流程.如圖2.1中Merge-Join操作符左右節(jié)點均設(shè)有Sort操作符與Rpc操作符,其中Rpc操作符負(fù)責(zé)向Storage請求數(shù)據(jù),Storage篩選符合過濾條件的數(shù)據(jù)并返回給Rpc操作符,而Sort操作符用于對Rpc操作符返回的數(shù)據(jù)進行排序,繼而在Merge Join操作符內(nèi)進行合并連接運算.

圖2 分布式j(luò)oin操作執(zhí)行計劃與基于Semi join算法優(yōu)化后的執(zhí)行計劃Fig.2Execution plan of distributed join operation and optimized execution plan based on Semi join algorithm

Merge-Join對左右表數(shù)據(jù)的請求是同時發(fā)送的,并且一般情況下對左表數(shù)據(jù)的請求會附帶充分的過濾條件,而對右表的數(shù)據(jù)請求往往攜帶較少的過濾條件,甚至沒有過濾條件.由以上的數(shù)據(jù)請求流程分析出分布式j(luò)oin的執(zhí)行計劃對右表數(shù)據(jù)過濾的不足,假設(shè)右表數(shù)據(jù)量很大或者過濾條件并不能有效地減少無用數(shù)據(jù)(不會產(chǎn)生連接結(jié)果的數(shù)據(jù))的傳輸,那么網(wǎng)絡(luò)傳輸開銷會成為系統(tǒng)的瓶頸.針對分布式j(luò)oin操作對右表數(shù)據(jù)過濾不足這一短板,本文第3節(jié)提出了基于Semi-Join算法的優(yōu)化方法.

3 基于Semi-Join算法的優(yōu)化方案

3.1 優(yōu)化后執(zhí)行流程

連接算法依然采用Merge-Join,R表與S表的連接條件為R.ID=S.ID,ID分別為兩表的主鍵.如圖2.2所示,采用Semi-Join算法的數(shù)據(jù)請求流程分為以下三個步驟:

①通過R表的過濾條件獲取R表結(jié)果集Result-set(R);

②將R表結(jié)果中的ID列作為S表的過濾條件(S.ID in(Result-Set(R.ID))),也就是使用get的方法連同S表原有的過濾條件一同過濾S表數(shù)據(jù);

③將過濾后的S表結(jié)果集Result-Set(S)與R表結(jié)果集Result-Set(R)進行合并連接運算.

以上這種通過主鍵定位get的方式來過濾S表數(shù)據(jù)的方法稱之為Semi-get-Join.此外,還有一種通過主鍵范圍來過濾數(shù)據(jù)的scan方法,稱之為Semi-range-Join.

3.2 主鍵范圍過濾

主鍵范圍過濾即通過R表的結(jié)果集Result-Set(R),如圖2.3所示,構(gòu)造關(guān)于S表主鍵ID的范圍,并且以這個主鍵范圍作為過濾條件篩選S表主鍵ID在此范圍內(nèi)的所有數(shù)據(jù). Semi-range-Join的數(shù)據(jù)請求流程分為以下三個步驟:

①通過R表的過濾條件獲取R表結(jié)果集Result-set(R);

②將R表結(jié)果中ID列的范圍(Range(MIN(R.ID),MAX(R.ID)))作為S表的過濾條件,也就是使用scan的方法并連同S表原有的過濾條件一同過濾S表數(shù)據(jù);

③將過濾后的S表結(jié)果集Result-Set(S)與R表結(jié)果集Result-Set(R)進行合并連接運算.

4 實驗分析

4.1 實驗系統(tǒng)原型

本文選擇OceanBase系統(tǒng)作為實驗原型,原因在于OceanBase系統(tǒng)的架構(gòu)符合第二節(jié)介紹的分布式系統(tǒng),并且其Join處理流程也滿足第二節(jié)介紹的分布式Join操作模型.

OceanBase中有四種Server:RootServer(RS)、UpdateServer(UPS)、ChunkServer(CS)、MergeServer(MS).RS是集群的主控節(jié)點,UPS與CS分別作為增量數(shù)據(jù)與基線數(shù)據(jù)的存儲與訪問節(jié)點,MS為查詢引擎.

因此本文在MS上實現(xiàn)了Semi-get-Join與Semi-range-Join兩種算法,并與OceanBase原有的Merge-Join在不同場景下進行響應(yīng)時間的對比,總結(jié)Semi-get-Join與Semi-range-Join各自的適應(yīng)場景.

4.2 實驗環(huán)境與參數(shù)

OceanBase集群配置:主控節(jié)點RS與增量數(shù)據(jù)存儲節(jié)點UPS共用一臺,基線數(shù)據(jù)存儲節(jié)點CS三臺,查詢引擎MS一臺用于接收SQL請求.所有服務(wù)器的配置如表1.

表1 集群服務(wù)器配置Tab.1Configuration of cluster server

數(shù)據(jù)集:使用sysbench提供的數(shù)據(jù)生成器生成數(shù)據(jù).共有5張表R,S1w,S10w,S100w, S1000w.每張表均有4列(ID,K,C,Pad),其中ID與K列數(shù)據(jù)類型為整型,C與Pad為字符型,ID為主鍵.其中,R表數(shù)據(jù)量為10萬,S1w、S10w、S100w、S1000w四張表的數(shù)據(jù)量分別為1萬、10萬、100萬、1 000萬.

連接關(guān)系與連接條件:連接關(guān)系為R??S,連接條件為R.ID=S.ID,并且R表有過濾條件R.ID6(100 or 1 000 or 10 000 or 100 000),以此來控制結(jié)果集大小,S表無任何過濾條件.

密度(Density):|Result-Set(R)|表示R表結(jié)果集中數(shù)據(jù)的行數(shù),|Range(MIN(R.ID), MAX(R.ID))|表示S表的ID列落在Range(MIN(R.ID),MAX(R.ID))范圍內(nèi)的數(shù)據(jù)的行數(shù).

表示S表理論上需要過濾的數(shù)據(jù)量與實際返回的數(shù)據(jù)量之比.如Density=0.1,即假設(shè)R表結(jié)果集行數(shù)為1 000,若經(jīng)過R表的ID列過濾后S表返回1 000行數(shù)據(jù),而實際上S表內(nèi)的ID列落在Range(MIN(R.ID),MAX(R.ID))這個范圍的數(shù)據(jù)為10000行,則Density為1 000/10 000,即為0.1.

4.3 實驗結(jié)果分析

圖3所示為在S表數(shù)據(jù)量分別為1萬行、10萬行、100萬行、1 000萬行的情況下,R表不同的結(jié)果集大小對Merge-join、Semi-get-Join以及Semi-range-Join算法響應(yīng)時間的影響.通過分析可以得出Merge-Join的響應(yīng)時間同時受到S表與R表數(shù)據(jù)量的影響,S表與R表的數(shù)據(jù)量越大,Merge-Join的響應(yīng)時間越長,原因在于Merge-Join并沒有針對S表的過濾進行優(yōu)化.而Semi-get-Join以及Semi-range-Join的響應(yīng)時間并不受S表數(shù)據(jù)量大小的影響,僅與R表的數(shù)據(jù)量有關(guān),R表數(shù)據(jù)量越大響應(yīng)時間越長.

圖3 Merge-Join、Semi-get-Join與Semi-range-Join響應(yīng)時間對比Fig.3Contrast of response time among Merge-Join、Semi-get-Join and Semi-range-Join

并且在S表數(shù)據(jù)量超過100萬行,R表數(shù)據(jù)量小于1 000行的情況下,Semi-get-Join與Semi-range-Join的響應(yīng)時間要優(yōu)于Merge-Join,并且Semi-range-Join的響應(yīng)時間還要優(yōu)于Semi-get-Join.

圖4顯示了在不同的密度下Semi-get-Join與Semi-range-Join的響應(yīng)時間對比,其中密度分別為0.1、0.01、0.001、0.000 1,并且S表的數(shù)據(jù)量均為1 000萬行,由于S表數(shù)據(jù)量有限,密度0.001與0.000 1中R表數(shù)據(jù)量的分類有所減少.通過這四種密度下Semi-get-Join與Semi-range-Join響應(yīng)時間的對比,可以看出當(dāng)密度小于0.01時,Semi-get-Join的響應(yīng)時間要優(yōu)于Semi-range-Join.

圖4 不同密度下Semi-get-Join與Semi-range-Join的響應(yīng)時間對比Fig.4Contrast of response time between Semi-get-Join and Semi-range-Join under the different densities

實驗結(jié)果表明,S表數(shù)據(jù)量超過100萬行,R表數(shù)據(jù)量小于1 000行的情況下,Semi-get-Join與Semi-range-Join的響應(yīng)時間要優(yōu)于Merge-Join.并且當(dāng)密度小于0.01時,Semi-get-Join的響應(yīng)時間要優(yōu)于Semi-range-Join.

5 總結(jié)

本文通過對分布式系統(tǒng)的Join操作的分析,基于Semi-Join提出了Semi-get-Join和Semirange-Join算法,并在OceanBase系統(tǒng)上做了相應(yīng)的實現(xiàn).實驗結(jié)果表明,在右表數(shù)據(jù)量大且左表通過過濾條件過濾后只有少量數(shù)據(jù)的場景下,這兩種算法能夠顯著提高Join操作的性能.同時,在小密度的場景下,Semi-get-join的性能要優(yōu)于Semi-range-Join.

［1］BERNSTEIN P A,CHIU D M W.Using semi-joins to solve relational queries[J].Journal of the ACM,1981, 28(1):25-40.

［2］AFRATI F N,ULLMAN J D.Optimizing multiway joins in a map-reduce environment[J].IEEE Transactions on Knowledge&Data Engineering,2011,23(9):1282-1298.

［3］BEAME P,KOUTRIS P,DAN S.Communication steps for parallel query processing[J].Computer Science,2013: 273-284.

［4］CHU S,BALAZINSKA M,SUCIU D.From theory to practice:Efficient join query evaluation in a parallel database system[C]//Proceedings of the 2015 ACM SIGMOD International Conference on Management of Data. ACM,2015:63-78.

［5］ELSEIDY M,ELGUINDY A,VITOROVIC A,et al.Scalable and adaptive online joins[J].Proceedings of the Vldb Endowment,2014,7(6):441-452.

［6］OKCAN A,RIEDEWALD M.Processing theta-joins using MapReduce[C]//ACM SIGMOD International Conference on Management of Data,SIGMOD 2011,Athens,Greece,June.2011:949-960.

［7］ZHANG X,CHEN L,WANG M.Efficient multi-way theta-join processing using mapreduce[J].Proceedings of the Vldb Endowment,2012,5(11):1184-1195.［8］SCHNEIDER D A,DEWITT D J.Tradeoffs in processing complex join queries via hashing in multiprocessor database machines[C]//International Conference on Very Large Data Bases,August 13-16,1990,Brisbane, Queensland,Australia.1990:469-480.

［9］NGO H Q,CHRISTOPHER,RUDRA A.Skew strikes back:new developments in the theory of join algorithms[J].AcmSigmod Record,2014,42(4):5-16.

(責(zé)任編輯:李萬會)

Implementation of Semi-Join algorithm in a distributed system

QIAN Zhao-ming,WANG Lei,YU Sheng-jun,GONG Xue-qing
(Institute for Data Science and Engineering,East China Normal University, Shanghai200062,China)

As the scope of application of the new distributed system is becoming wider, the application is no longer satisfied with using primary key access to read the data,and how to efficiently achieve such complex operations as Join in these systems has become a research hot topic.This paper introduces how to realize the Join operation in the distributed systems based on the Semi-Join algorithm,and puts forward two ways to get the data in right table,and the performance of the algorithm is also analyzed through experiments.

distributed database;Join operation;Semi-Join algorithm

TP301.6

A

10.3969/j.issn.1000-5641.2016.05.009

1000-5641(2016)05-0075-06

2016-05

國家自然科學(xué)基金(61332006);國家863計劃項目(2015AA015307)

錢招明,男,碩士研究生,研究方向為分布式數(shù)據(jù)庫.E-mail:51141500029@ecnu.cn.