基于Spark的大數(shù)據(jù)統(tǒng)計(jì)中等值連接問題的優(yōu)化

劉容辰，周明強(qiáng)，皮興杰，趙欣

（重慶大學(xué)計(jì)算機(jī)學(xué)院，重慶 400044）

基于Spark的大數(shù)據(jù)統(tǒng)計(jì)中等值連接問題的優(yōu)化

劉容辰，周明強(qiáng)，皮興杰，趙欣

（重慶大學(xué)計(jì)算機(jī)學(xué)院，重慶 400044）

伴隨著互聯(lián)網(wǎng)應(yīng)用技術(shù)的飛速發(fā)展，導(dǎo)致傳統(tǒng)的數(shù)據(jù)處理技術(shù)已經(jīng)無(wú)法滿足對(duì)大數(shù)據(jù)高效處理的要求。因此對(duì)現(xiàn)有的大數(shù)據(jù)的統(tǒng)計(jì)分析便急需相應(yīng)的大數(shù)據(jù)技術(shù)的支持。為了解決實(shí)際Spark應(yīng)用中的Join操作低效的問題，首先，提出一種高效的基于BloomFilter過(guò)濾再分區(qū)算法，通過(guò)該算法率先過(guò)濾掉絕大部分不符合條件的無(wú)效連接，然后針對(duì)過(guò)濾數(shù)據(jù)產(chǎn)生的傾斜問題進(jìn)行再分區(qū)操作，以便能充分發(fā)揮各個(gè)工作節(jié)點(diǎn)的計(jì)算資源，達(dá)到在最大程序上優(yōu)化Join過(guò)程的目的。

大數(shù)據(jù)；Spark；等值連接；BloomFilter；Shuffle

1 相關(guān)工作

近些年來(lái)，伴隨著互聯(lián)網(wǎng)應(yīng)用技術(shù)的飛速發(fā)展，導(dǎo)致數(shù)據(jù)量巨大且種類繁多的數(shù)據(jù)給傳統(tǒng)的數(shù)據(jù)處理技術(shù)帶來(lái)了巨大的挑戰(zhàn)，因此大數(shù)據(jù)便成為現(xiàn)在階段最受到關(guān)注且有待解決的熱門問題[1，4]。在傳統(tǒng)數(shù)據(jù)庫(kù)（如：Oracle、MySQL）中，Join操作是特別常見且耗時(shí)的操作[2]。如何優(yōu)化這種等值連接操作便成為了大家的研究重點(diǎn)[3,5]。

文獻(xiàn)[6]提出了一種大表等值連接的優(yōu)化算法，其優(yōu)化思想是利用Bit-Map壓縮算法對(duì)大表的連接屬性進(jìn)行壓縮處理，這種優(yōu)化算法可以提前過(guò)濾掉一些不符合連接條件的記錄，減少Shuffle階段的數(shù)據(jù)量。但是這種優(yōu)化算法在數(shù)據(jù)過(guò)濾階段的誤判率特別高，因此會(huì)誤判很多不符合連接條件的數(shù)據(jù)，Shuffle階段的數(shù)據(jù)量仍然會(huì)比較大，數(shù)據(jù)過(guò)濾效果并不明顯。文獻(xiàn)[7]的優(yōu)化思路同樣是在大表相連以前過(guò)濾掉一些不符合條件的記錄，其過(guò)濾算法采用BloomFilter數(shù)據(jù)結(jié)構(gòu)，因?yàn)锽loomFilter結(jié)構(gòu)相對(duì)于Bit-Map而言，它的數(shù)據(jù)過(guò)濾效果更好，但它的缺點(diǎn)是生成的位數(shù)組更大。文獻(xiàn)[8]結(jié)合了BloomFilter的思想，提出一種高效的基于BloomFilter數(shù)據(jù)結(jié)構(gòu)的大表等值連接優(yōu)化算法，但是針對(duì)RDD過(guò)濾后產(chǎn)生的數(shù)據(jù)傾斜的問題卻沒有進(jìn)一步的優(yōu)化操作。

本文借鑒了傳統(tǒng)數(shù)據(jù)庫(kù)的統(tǒng)計(jì)查詢方法以及基于Hadoop平臺(tái)的統(tǒng)計(jì)思想，在數(shù)據(jù)統(tǒng)計(jì)過(guò)程中，優(yōu)化了數(shù)據(jù)的等值連接過(guò)程，針對(duì)RDD進(jìn)行預(yù)先數(shù)據(jù)過(guò)濾產(chǎn)生數(shù)據(jù)傾斜的問題，借鑒一致性的思想，進(jìn)一步對(duì)過(guò)濾后RDD進(jìn)行重分區(qū)操作，以便能充分發(fā)揮各個(gè)工作節(jié)點(diǎn)的計(jì)算資源，達(dá)到在最大程序上優(yōu)化Join過(guò)程的目的，并通過(guò)實(shí)驗(yàn)驗(yàn)證了該算法的有效性。

2 Bloom Filter介紹

BloomFilter過(guò)濾器是一種具備極高的空間效率的數(shù)據(jù)結(jié)構(gòu)，主要用于檢索一個(gè)元素是否已經(jīng)存在于某一個(gè)數(shù)據(jù)集合中[9]。如圖1所示，x，y，z這三個(gè)哈希函數(shù)所對(duì)應(yīng)數(shù)值均為1，都是符合條件的數(shù)據(jù)，而w則不是。

BloomFilter的優(yōu)點(diǎn)是空間效率高和查詢時(shí)間較短，其缺點(diǎn)則是會(huì)存在一定的誤判率。對(duì)于具備n條記錄的數(shù)據(jù)集，k個(gè)哈希函數(shù)，位數(shù)組大小為m，造成的誤判率p如下：

圖1 BloomFilter過(guò)濾示意圖

本文將BloomFilter應(yīng)用在Spark平臺(tái)的Join操作中，以達(dá)到提前過(guò)濾掉不符合連接條件的數(shù)據(jù)的目的。

3 過(guò)濾再分區(qū)的大表等值連接算法介紹

3.1 過(guò)濾算法介紹

過(guò)濾算法主要可以分成兩個(gè)部分。第一部分：對(duì)抽取的兩張數(shù)據(jù)表的連接屬性進(jìn)行去重操作，然后對(duì)經(jīng)過(guò)去重后操作后的連接屬性進(jìn)行BloomFilter壓縮處理，從而得到過(guò)濾所需要的位數(shù)組數(shù)據(jù)。對(duì)兩個(gè)位數(shù)組進(jìn)行與運(yùn)算便可得到最終的位數(shù)組（BF）。第二部分，使用最終生成的BF位數(shù)組對(duì)待Join的兩張數(shù)據(jù)表進(jìn)行過(guò)濾，提前過(guò)濾掉所有不滿足連接條件的記錄，當(dāng)經(jīng)過(guò)Judge運(yùn)算之后，其返回false的元素將被提前過(guò)濾掉。Judge操作如下：首先針對(duì)每條記錄數(shù)據(jù)的連接屬性，使用生成的位數(shù)組的k個(gè)哈希函數(shù)，計(jì)算出k個(gè)Hash值，然后再對(duì)m取模運(yùn)算，從而得到k個(gè)0～m-1的值，最后檢查BF位數(shù)組中k個(gè)位置的值是否均為1，BF位數(shù)組中這個(gè)k個(gè)位置值若全為1，則返回true，反之則，返回false[4]。過(guò)濾算法整體步驟如圖2所示。

圖2 過(guò)濾算法整體示意圖

在上述兩個(gè)過(guò)程中，兩個(gè)數(shù)據(jù)表一定要使用相同的Hash函數(shù)進(jìn)行壓縮處理。最后利用Broadcast操作將該位數(shù)組廣播到集群的其他工作節(jié)點(diǎn)上，從而加速數(shù)據(jù)表的過(guò)濾階段。

3.2 分區(qū)策略介紹

（1）首先，根據(jù)table_a_p，table_b_p的元組數(shù)量，從而計(jì)算連接并行度pd，pd=min（table_a_p.partionSize+ table_b_p.partionSize，w），其中partionSize表示分區(qū)數(shù)，w表示用于控制并行度的最大值。

（2）分別選取table_a_p，table_b_p的連接屬性key，根據(jù)設(shè)定好的采樣率ω進(jìn)行對(duì)兩者的并集采樣，得到樣本sampleKey。

（3）然后，計(jì)算sampleKey中key的Hash值，根據(jù)（1）計(jì)算的pd值，將Hash空間劃分成為pd個(gè)區(qū)間，并且需要保證每個(gè)區(qū)間的樣本大小相等。

（4）最后，將table_a_f，table_b_f按照步驟（3）中劃分出區(qū)間，通過(guò)Hash分配到集群對(duì)應(yīng)節(jié)點(diǎn)上。

對(duì)經(jīng)過(guò)上述的過(guò)濾在分區(qū)操作后數(shù)據(jù)表執(zhí)行普通的Join操作，再將處理過(guò)后的RDD轉(zhuǎn)換成數(shù)據(jù)表，然后將相關(guān)業(yè)務(wù)sql進(jìn)行HashJoin操作，最后，將生成結(jié)果保存到傳統(tǒng)數(shù)據(jù)庫(kù)（如：Oracle,MySQL）中。

3.3 過(guò)濾再分區(qū)算法分析

本文所提出的基于過(guò)濾再分區(qū)的大表等值連接算法——BF-P-Join（BloomFilter-Partitioner-Join）主要包含三個(gè)主要階段及數(shù)據(jù)表過(guò)濾階段、數(shù)據(jù)表再分區(qū)階段和最后的HashJoin階段。Spark處理等值連接的方法主要有兩種：BroadcastJoin方式和HashJoin方式。本節(jié)將從代價(jià)開銷和適用場(chǎng)景兩個(gè)角度對(duì)比這幾種處理方式優(yōu)劣。

完成數(shù)據(jù)處理任務(wù)的開銷主要在Shuffle階段的網(wǎng)絡(luò)通信開銷和Join階段的執(zhí)行時(shí)間。代價(jià)分析中的符號(hào)描述如表1所示：

（1）HashJoin方式，對(duì)原數(shù)據(jù)表直接進(jìn)行關(guān)聯(lián)，在整個(gè)關(guān)聯(lián)過(guò)程中的重分區(qū)操作會(huì)出現(xiàn)Shuffle操作，因此這種方式的網(wǎng)絡(luò)通信的開銷主要與數(shù)據(jù)表的數(shù)據(jù)量相關(guān)：

（2）BroadcastJoin方式，將其中一張關(guān)聯(lián)表以廣播的方式發(fā)送到集群的其他節(jié)點(diǎn)上，連接過(guò)程中無(wú)Shuf-fle操作，因此網(wǎng)絡(luò)開銷主要與廣播表的數(shù)據(jù)量相關(guān)。通常選擇數(shù)據(jù)量較小的數(shù)據(jù)表作為廣播表，假設(shè)table_b是廣播表，則它的網(wǎng)絡(luò)開銷如下：

BF-P-Join方式，其開銷主要由兩部分構(gòu)成。第一個(gè)部分的開銷主要在位數(shù)組生成階段。第二部分是數(shù)據(jù)表執(zhí)行HashJoin操作時(shí)Shuffle操作產(chǎn)生的數(shù)據(jù)量所產(chǎn)生的網(wǎng)絡(luò)開銷。總代價(jià)開銷如下：

從整體上看，BF-P-Join方式雖然增加了分區(qū)階段的時(shí)間開銷，但在最后的運(yùn)行階段可以充分利用了集群中各節(jié)點(diǎn)的并行運(yùn)算性能，從而縮短統(tǒng)計(jì)時(shí)間。

表1 代價(jià)分析符號(hào)描述

4 實(shí)驗(yàn)

將本文提出的基于過(guò)濾再分區(qū)的大表等值連接算法，與SparkSql默認(rèn)的HashJoin算法以及文獻(xiàn)[4]中提出的BF-Join算法進(jìn)行對(duì)比實(shí)驗(yàn)，以驗(yàn)證該算法在Shuffle階段的數(shù)據(jù)讀寫量及算法的整體運(yùn)行時(shí)間要更加高效。本次實(shí)驗(yàn)選取了三組不同規(guī)模的數(shù)據(jù)集，如表2所示。

表2 數(shù)據(jù)集大小

為了使得實(shí)驗(yàn)結(jié)果更加可信，在實(shí)驗(yàn)過(guò)程中對(duì)每組實(shí)驗(yàn)重復(fù)執(zhí)行一百次，并記錄實(shí)驗(yàn)結(jié)果的平均值。實(shí)驗(yàn)結(jié)果如表3、表4和圖3所示。

根據(jù)以上實(shí)驗(yàn)結(jié)果，可以看出，在Shuffle讀和Shuffle寫兩個(gè)方面，BF-Join算法和BF-P-Join算法都要優(yōu)于HashJoin算法。BF-P-Join算法需要因?yàn)檫M(jìn)行過(guò)濾后再分區(qū)的階段，該階段同樣也會(huì)產(chǎn)Shuffle操作，因此這兩方面上不如BF-Join算法。從算法整體運(yùn)行時(shí)間方面，可以看出BF-Join算法和BF-P-Join算法同樣優(yōu)于HashJoin算法。BF-P-Join算法因?yàn)槌隽艘徊椒謪^(qū)操作，在數(shù)據(jù)量較小的情況下，集群的任務(wù)量較少，沒有出現(xiàn)數(shù)據(jù)傾斜的現(xiàn)象，從而導(dǎo)致整體運(yùn)行時(shí)間略慢于BF-Join算法。在數(shù)據(jù)量較大的情況下，可以看出上本文提出的BF-P-Join算法明顯優(yōu)于BF-Join算法。

表3 Shuffle寫數(shù)據(jù)量對(duì)比

表4 Shuffle讀數(shù)據(jù)量對(duì)比

圖3 算法整體運(yùn)行時(shí)間對(duì)比圖

5 結(jié)語(yǔ)

本文針對(duì)大表的等值連接問題，提出了基于過(guò)濾再分區(qū)的BFP算法。該算法在大表等值連接過(guò)程中，首先采用BloomFilter結(jié)構(gòu)過(guò)濾大部分不符合連接條件的數(shù)據(jù)。然后，通過(guò)自定義分區(qū)操作將剩余的計(jì)算任務(wù)平均分配到集群中的各計(jì)算節(jié)點(diǎn)上，以達(dá)到優(yōu)化Join過(guò)程的目的。最后，通過(guò)對(duì)比實(shí)驗(yàn)驗(yàn)證了該算法的有效性。

參考文獻(xiàn)：

[1]沈曉磊.基于“大數(shù)據(jù)”的重點(diǎn)人員管控系統(tǒng)的設(shè)計(jì)與實(shí)現(xiàn)[D].蘇州大學(xué)，2014.

[2]馬建光,姜巍.大數(shù)據(jù)的概念、特征及其應(yīng)用[J].國(guó)防科技大學(xué)，2013,34（2）:10-17.

[3]王妍,柴劍平.大數(shù)據(jù)及相關(guān)技術(shù)解讀[J].廣播電視信息，2014（2）:18-21.

[4]張昕.大數(shù)據(jù)環(huán)境下電子商務(wù)的發(fā)展探析[J].中國(guó)科技信息，2014（23）:197-200.

[5]陶雪嬌,胡曉峰,劉洋.大數(shù)據(jù)研究綜述[J].系統(tǒng)仿真學(xué)報(bào)，2013（S1）:142-146.

[6]孫惠.基于Hadoop框架的大數(shù)據(jù)集連接優(yōu)化算法[D].南京郵電大學(xué),2013.

[7]Ramesh S,Papapertrou O.Optimizing Distributed Joinswith Bloom Filters[J].Lecture Notes in Computer Science,2008,5375:145-156.

[8]周思偉.Spark大表等值連接的優(yōu)化及其在網(wǎng)絡(luò)流量數(shù)據(jù)分析的應(yīng)用研究[D].華南理工大學(xué),2015.

[9]王鳳君.P2P存儲(chǔ)系統(tǒng)中副本管理策略的研究[D].北京郵電大學(xué)，2011.

Optim ization of the Equi-Join Problem Based on Big Data in Spark

LIU Rong-chen，ZHOU Ming-qiang，PIXing-jie，ZHAO Xin
（College of Computer Science,Chongqing University,Chongqing 400044）

With the rapid development of Internet application technology,leading to the traditional data processing technology has been unable to meet the requirements for the efficient processing of large data.So the existing large-scale statistical analysis of the big data will be in urgentneed of the support of the corresponding large data technology.In order to solve the problem of the joining operation inefficient in practical Spark application,firstly,proposes an efficientalgorithm of BloomFilter filter re-partitioning,uses the algorithm to filter outmost of the invalid connections that do notmeet the criteria.And then，re-partition the filtering data which is tilted,in order tomake full use of the computing resources various nodes to achieve themaximum process to optimize the purpose of the join process.

Big Data;Spark;Equi-Join;BloomFilter;Shuffle

1007-1423（2017）12-0003-04

10.3969/j.issn.1007-1423.2017.12.001

劉容辰（1992-），男，湖南懷化人，碩士，研究方向?yàn)閺?fù)雜網(wǎng)絡(luò)、數(shù)據(jù)挖掘

周明強(qiáng)（1977-），男，重慶人，博士學(xué)位，研究方向?yàn)閺?fù)雜網(wǎng)絡(luò)、數(shù)據(jù)挖掘

皮興杰（1989-），男，湖北襄陽(yáng)人，碩士，研究方向?yàn)閺?fù)雜網(wǎng)絡(luò)、數(shù)據(jù)挖掘

趙欣（1991-），男，甘肅天水人，碩士，研究方向?yàn)閿?shù)據(jù)挖掘

2017-03-06

2017-04-20