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

錢招明,王雷,余晟雋,宮學慶

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

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

錢招明,王雷,余晟雋,宮學慶

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

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

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

0 引言

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

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

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

1 相關工作

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

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

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

2.1 分布式系統(tǒng)

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

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

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

Storage一般提供兩種數(shù)據(jù)訪問方式:讀操作或寫操作.通常,Storage可提供讀操作也可提供寫操作,對于讀寫分離的分布式系統(tǒng)而言,Storage被分為讀節(jié)點與寫節(jié)點兩種角色.而特別針對讀操作,并且在通過主鍵訪問數(shù)據(jù)的場景下,本文考慮兩種不同的數(shù)據(jù)過濾方法:一是通過主鍵定位數(shù)據(jù)的Get方法;二是經由主鍵范圍掃描數(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é)點均設有Sort操作符與Rpc操作符,其中Rpc操作符負責向Storage請求數(shù)據(jù),Storage篩選符合過濾條件的數(shù)據(jù)并返回給Rpc操作符,而Sort操作符用于對Rpc操作符返回的數(shù)據(jù)進行排序,繼而在Merge Join操作符內進行合并連接運算.

圖2 分布式join操作執(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ù)請求流程分析出分布式join的執(zhí)行計劃對右表數(shù)據(jù)過濾的不足,假設右表數(shù)據(jù)量很大或者過濾條件并不能有效地減少無用數(shù)據(jù)(不會產生連接結果的數(shù)據(jù))的傳輸,那么網絡傳輸開銷會成為系統(tǒng)的瓶頸.針對分布式join操作對右表數(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表結果集Result-set(R);

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

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

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

3.2 主鍵范圍過濾

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

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

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

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

4 實驗分析

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

本文選擇OceanBase系統(tǒng)作為實驗原型,原因在于OceanBase系統(tǒng)的架構符合第二節(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在不同場景下進行響應時間的對比,總結Semi-get-Join與Semi-range-Join各自的適應場景.

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

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

表1 集群服務器配置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萬.

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

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

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

4.3 實驗結果分析

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

圖3 Merge-Join、Semi-get-Join與Semi-range-Join響應時間對比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ōu)于Merge-Join,并且Semi-range-Join的響應時間還要優(yōu)于Semi-get-Join.

圖4顯示了在不同的密度下Semi-get-Join與Semi-range-Join的響應時間對比,其中密度分別為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響應時間的對比,可以看出當密度小于0.01時,Semi-get-Join的響應時間要優(yōu)于Semi-range-Join.

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

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

5 總結

本文通過對分布式系統(tǒng)的Join操作的分析,基于Semi-Join提出了Semi-get-Join和Semirange-Join算法,并在OceanBase系統(tǒng)上做了相應的實現(xiàn).實驗結果表明,在右表數(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.

(責任編輯:李萬會)

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

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

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