流數(shù)據(jù)實(shí)時(shí)接收方案的研究

張笑燕，劉志浩，杜曉峰，陸天波

（北京郵電大學(xué)計(jì)算機(jī)學(xué)院（國家示范性軟件學(xué)院），北京 100876）

0 引言

隨著信息社會(huì)飛速發(fā)展，數(shù)據(jù)產(chǎn)生的速度越來越快?，F(xiàn)實(shí)中存在一類常見的業(yè)務(wù)，即將大量源源不斷到達(dá)的流數(shù)據(jù)S先與存儲(chǔ)在磁盤上的關(guān)系表R進(jìn)行連接，對S進(jìn)行半流連接關(guān)聯(lián)更新操作（如去重、修正等）后再將其寫入數(shù)據(jù)倉庫[1]。該過程被定義為S∞CR[2]。其中，C代表某種半流連接操作。

由于源數(shù)據(jù)系統(tǒng)的不同，邏輯上相同的元組可能具有不同的值，在數(shù)據(jù)進(jìn)入數(shù)據(jù)倉庫時(shí)，需要通過關(guān)聯(lián)表進(jìn)行統(tǒng)一，保證數(shù)據(jù)的一致性。

為了更具體地理解這種半流連接關(guān)聯(lián)更新操作，圖1 展示了數(shù)據(jù)倉庫中涉及該操作的一個(gè)例子。由于數(shù)據(jù)來源的不同或數(shù)據(jù)的延遲到達(dá)，系統(tǒng)中同一個(gè)id 可能具有不同的名稱，在數(shù)據(jù)進(jìn)入數(shù)據(jù)倉庫之前，需要使用一張關(guān)系表將id 轉(zhuǎn)換為系統(tǒng)內(nèi)部的id，對不同的名稱進(jìn)行統(tǒng)一。

圖1 數(shù)據(jù)倉庫關(guān)聯(lián)更新示意

在真實(shí)的數(shù)據(jù)倉庫系統(tǒng)中，R保存在磁盤上，一般占用空間較大，無法全部放入內(nèi)存；S包含不斷到達(dá)的流數(shù)據(jù)，其中的每個(gè)元組都需要和R進(jìn)行連接操作[3-5]。這便產(chǎn)生了一個(gè)問題：S中的元組以流數(shù)據(jù)的形式不斷快速到達(dá)，而R中的元組需要通過磁盤I/O 以相對較低的速率讀取，造成更新時(shí)需要消耗大量時(shí)間等待磁盤I/O，導(dǎo)致數(shù)據(jù)處理的延遲。

事實(shí)上，為了解決流數(shù)據(jù)連接操作中的各種問題，與此有關(guān)的研究從未中斷。Polyzotis 等[6]提出了MESHJOIN 算法，通過對關(guān)系表的分組讀取和對流數(shù)據(jù)的分頁讀取，把對磁盤的I/O 時(shí)間分?jǐn)偟搅巳舾蓚€(gè)流數(shù)據(jù)元組的讀取中，平衡了關(guān)系表的讀取速率和流數(shù)據(jù)的到達(dá)速率的差異。但是MESHJOIN 算法并沒有考慮流數(shù)據(jù)S的特征和關(guān)系表R的組織，因此在處理傾斜數(shù)據(jù)時(shí)表現(xiàn)較差[7]。Vaidehi 等[8]提出了分布式嵌套循環(huán)連接處理（DNLJP,distributed nested loop join processing）算法，對流數(shù)據(jù)的來源進(jìn)行分組，根據(jù)其所定義的不同查詢成本、查詢時(shí)間等對連接查詢操作進(jìn)行優(yōu)化，降低了分布式集群中各節(jié)點(diǎn)之間通信的成本，提高了對海量數(shù)據(jù)連接查詢的效率。DNLJP 是可以用于分布式集群的算法，而 MESHJOIN 和CACHEJOIN 算法只適于單機(jī)環(huán)境，無法直接應(yīng)用于分布式環(huán)境[9-10]。Naeem 等[11]提出 的CACHEJOIN 算法考慮了數(shù)據(jù)傾斜的問題，通過引入一個(gè)緩存，將關(guān)系表中較常用的部分存儲(chǔ)在緩存中，其中的元組有較高的概率和流數(shù)據(jù)匹配成功，減少了磁盤的I/O 次數(shù)。Jeon 等[12]提出了DS-join方案，通過使用幾種流處理引擎的微批處理模式，控制數(shù)據(jù)分區(qū)，減少了各節(jié)點(diǎn)之間的網(wǎng)絡(luò)通信頻率，同時(shí)引入緩存模塊對連接操作并行處理，減少磁盤I/O 次數(shù)，并且擁有自動(dòng)調(diào)整緩存區(qū)大小的優(yōu)化設(shè)置。為了保證分布式流連接操作中處理結(jié)果具有完整性和一致性，Yuan 等[13]測試了數(shù)據(jù)倉庫接收到的流數(shù)據(jù)出現(xiàn)錯(cuò)誤時(shí)連接操作結(jié)果的質(zhì)量優(yōu)劣，提出了基于有序傳播模型的、應(yīng)用于分布式流連接系統(tǒng)的Eunomia 方法，保證了所有連接器的元組到達(dá)順序的一致性，能夠消除數(shù)據(jù)傳播過程中的某些錯(cuò)誤，提高了分布式流連接處理的擴(kuò)展性和吞吐率。狄程等[14]設(shè)計(jì)并實(shí)現(xiàn)了對多源異構(gòu)流數(shù)據(jù)的處理系統(tǒng)，在一定程度上消除了數(shù)據(jù)的結(jié)構(gòu)不同對數(shù)據(jù)引入、連接處理造成的復(fù)雜問題，使流處理服務(wù)化、模塊化，減少了流數(shù)據(jù)處理程序的開發(fā)成本。

因此，本文在已有研究的基礎(chǔ)上進(jìn)行改進(jìn)和優(yōu)化，提出了應(yīng)用于分布式環(huán)境的D-CACHEJOIN 算法。D-CACHEJOIN 算法的關(guān)鍵在于采用一致性哈希函數(shù)[15]的策略，將R分區(qū)存儲(chǔ)在不同節(jié)點(diǎn)，并在后續(xù)的匹配過程中使用這一計(jì)算結(jié)果，達(dá)到快速將R與S進(jìn)行匹配的目的。同時(shí)，本文使用服從Zipfian 分布的模擬數(shù)據(jù)[16-17]，定量計(jì)算算法執(zhí)行的成本開銷，以優(yōu)化算法的執(zhí)行效率[18]。實(shí)驗(yàn)表明，在擁有一定的可用內(nèi)存時(shí)，D-CACHEJOIN 算法在處理流數(shù)據(jù)時(shí)具有良好的實(shí)時(shí)處理性能，同時(shí)具有易于擴(kuò)展的特性。

1 基于緩存的分布式D-CACHEJOIN 算法

1.1 D-CACHEJOIN 算法的原理描述

與CACHEJOIN 算法類似，D-CACHEJOIN 算法擁有2 個(gè)階段——流檢測階段和磁盤檢測階段。流檢測階段使用流數(shù)據(jù)S作為輸入，磁盤檢測階段使用關(guān)系表R作為輸入，由于內(nèi)存大小的限制，兩階段每次都分別只處理S和R的一部分。

D-CACHEJOIN 算法執(zhí)行架構(gòu)如圖2 所示[11]，關(guān)系表R和流數(shù)據(jù)S是外部輸入，HS用于分次存儲(chǔ)S中的元組，實(shí)際中占用較大的空間；HR是緩存模塊，用于存儲(chǔ)R中匹配頻繁的元組[19]。

圖2 D-CACHEJOIN 算法執(zhí)行架構(gòu)

下面對本文算法的執(zhí)行架構(gòu)進(jìn)行說明。算法在流檢測階段，讀取流緩沖區(qū)的數(shù)據(jù)元組并與HR中存儲(chǔ)的高頻訪問元組進(jìn)行連接匹配，如果在HR中找到所需的元組，則算法生成該元組作為輸出，只有在無法匹配時(shí)，才會(huì)將該元組存儲(chǔ)到HS中并將其指針加入隊(duì)列，以便進(jìn)行后續(xù)的匹配操作。當(dāng)HS滿或S中已無數(shù)據(jù)元組時(shí)，流檢測階段結(jié)束，磁盤檢測階段開始。在磁盤檢測階段，算法將R中的部分元組讀取到內(nèi)存，得以分?jǐn)傁馁Y源較多的磁盤I/O 操作。讀取完成后，算法會(huì)將其與HS中的元組進(jìn)行匹配，匹配成功后，算法生成連接后的元組作為輸出。與此同時(shí)，匹配次數(shù)超過閾值的元組會(huì)被緩存到HR中用于流檢測。經(jīng)歷數(shù)次迭代后，HS中最先進(jìn)入的元組已經(jīng)完成了和R中所有元組的匹配，算法會(huì)將其刪除，此時(shí)HS中有了新的空閑位置，可以繼續(xù)存儲(chǔ)S中待匹配的元組，因此算法接下來再次切換到流檢測階段。

D-CACHEJOIN 算法的偽代碼實(shí)現(xiàn)如算法1所示[11]，其中w為流檢測階段讀取S的元組數(shù)，b為磁盤檢測階段讀取R的元組數(shù)。

算法1D-CACHEJOIN

20) 刪除HS中匹配完成的元組及其隊(duì)列中對應(yīng)的指針

21) end while

為了完整地處理流數(shù)據(jù)，算法的外層是無限循環(huán)，主要包括2 個(gè)部分，流檢測階段和磁盤檢測階段，二者交替進(jìn)行。

在步驟2)～步驟9)的流檢測階段，算法在流緩沖區(qū)讀取w個(gè)元組，對w中的每個(gè)元組t，算法首先檢測檢測t是否在HR中，如果是則將其連接并輸出結(jié)果；否則將t加入HS中，同時(shí)在隊(duì)列中加入t的指針。

在步驟10)～步驟20)的流檢測階段，算法在磁盤緩沖區(qū)讀取b個(gè)元組，對b中的每個(gè)元組r，算法首先檢測r是否在HS中，如果是則將其連接并輸出結(jié)果；否則將忽略該元組。由于HS需要存儲(chǔ)流中的元組，因此HS是一個(gè)多映射，r可能會(huì)成功匹配到HS中的多個(gè)元組，記該成功匹配的次數(shù)為f；如果某個(gè)r對應(yīng)的f大于閾值threshold Value，就將r加入HR中，如果HR已滿，則先隨機(jī)刪除HR中的一個(gè)元組，然后再將r加入HR。在每一次的迭代中，如果緩存HR未滿，說明閾值設(shè)置過高，此時(shí)可以自動(dòng)降低閾值；如果過于頻繁地從HR中刪除元組，說明閾值設(shè)置過低，此時(shí)可以自動(dòng)提高閾值。

1.2 D-CACHEJOIN 算法的系統(tǒng)實(shí)現(xiàn)

1.1 節(jié)說明了本文算法在單節(jié)點(diǎn)環(huán)境中的一般過程，本節(jié)將算法應(yīng)用到分布式環(huán)境中。為了敘述簡潔，本節(jié)以擁有2 個(gè)節(jié)點(diǎn)的數(shù)據(jù)倉庫為例。

數(shù)據(jù)倉庫接收從數(shù)據(jù)源經(jīng)網(wǎng)絡(luò)傳輸?shù)牧鲾?shù)據(jù)時(shí)，數(shù)據(jù)會(huì)隨機(jī)地分配到各節(jié)點(diǎn)上。數(shù)據(jù)向各節(jié)點(diǎn)的隨機(jī)分配是為了實(shí)現(xiàn)集群整體上各節(jié)點(diǎn)抽取-轉(zhuǎn)換-加載（ETL,extract-transform-load）的負(fù)載均衡，在實(shí)際處理中，可以在隨機(jī)選取第一個(gè)節(jié)點(diǎn)后，按照某個(gè)固定的順序依次選取各個(gè)節(jié)點(diǎn)作為數(shù)據(jù)的接收方，消除因選取節(jié)點(diǎn)而產(chǎn)生的額外的CPU 計(jì)算開銷。節(jié)點(diǎn)在接收到數(shù)據(jù)元組后，直接通過一致性哈希函數(shù)計(jì)算，如果元組屬于該節(jié)點(diǎn)，就將其保留，否則將元組轉(zhuǎn)發(fā)到對應(yīng)的節(jié)點(diǎn)，同時(shí)將計(jì)算得到的哈希值一并發(fā)送，避免后續(xù)產(chǎn)生重復(fù)的計(jì)算。D-CACHEJOIN 算法的整體架構(gòu)流程如圖 3 所示。對于 MESHJOIN 算法和CACHEJOIN 算法，由于其相當(dāng)于只在圖3 中的“連接”環(huán)節(jié)發(fā)揮作用，而數(shù)據(jù)元組在分布式環(huán)境中的接收、轉(zhuǎn)發(fā)等都在此之前就已經(jīng)完成，因此可以相對簡單地一并擴(kuò)展到分布式環(huán)境之中。

在基于本文算法的處理流程中，流數(shù)據(jù)S、關(guān)系表R和數(shù)據(jù)倉庫集群中各節(jié)點(diǎn)均采用同一個(gè)哈希函數(shù)進(jìn)行映射，有利于集群的擴(kuò)展。同時(shí)對于節(jié)點(diǎn)接收到的不屬于自身、需要轉(zhuǎn)發(fā)的元組，通過計(jì)算得到的哈希值將會(huì)隨元組一同轉(zhuǎn)發(fā)，避免了重復(fù)的計(jì)算。

從圖3 可以看出，數(shù)據(jù)倉庫將流數(shù)據(jù)隨機(jī)地依次分配給集群中的任一節(jié)點(diǎn)，從宏觀上看，集群中的所有節(jié)點(diǎn)在某一時(shí)間段內(nèi)同時(shí)讀取數(shù)據(jù)并進(jìn)行計(jì)算，保留屬于自身的數(shù)據(jù)、轉(zhuǎn)發(fā)屬于其他節(jié)點(diǎn)的數(shù)據(jù)。這樣的設(shè)計(jì)沒有屬于中心地位的節(jié)點(diǎn)，發(fā)揮了分布式集群并行處理數(shù)據(jù)的優(yōu)勢。數(shù)據(jù)不需要首先經(jīng)過中心節(jié)點(diǎn)的處理，而是直接在各節(jié)點(diǎn)之間進(jìn)行流轉(zhuǎn)，避免了由于存在中心節(jié)點(diǎn)導(dǎo)致系統(tǒng)出現(xiàn)性能瓶頸的問題。

圖3 D-CACHEJOIN 算法的整體架構(gòu)流程

2 D-CACHEJOIN 算法實(shí)驗(yàn)分析

2.1 性能評估模型

對于數(shù)據(jù)倉庫的ETL 過程而言，主要關(guān)注的性能指標(biāo)包括系統(tǒng)吞吐率、內(nèi)存消耗等。本文研究中使用的分布式系統(tǒng)中，各個(gè)節(jié)點(diǎn)的作用主要是參與計(jì)算和數(shù)據(jù)存儲(chǔ)等，大量運(yùn)算如備份、保證數(shù)據(jù)一致性等只在內(nèi)部進(jìn)行，對內(nèi)提供支持，對外只提供將多節(jié)點(diǎn)集群環(huán)境虛擬為單節(jié)點(diǎn)環(huán)境的接口，集群可以視為一個(gè)單節(jié)點(diǎn)環(huán)境。引入分布式系統(tǒng)直接導(dǎo)致算法增加的開銷是一致性哈希函數(shù)計(jì)算的開銷，為了實(shí)現(xiàn)負(fù)載均衡，令各節(jié)點(diǎn)隨機(jī)接收流數(shù)據(jù)，然后相互轉(zhuǎn)發(fā)數(shù)據(jù)。文獻(xiàn)[6]中已經(jīng)給出MESHJOIN 算法的成本模型，本文參考該模型，建立D-CACHEJOIN 算法的性能評估模型，模型參數(shù)如表1 所示。

表1 算法性能評估模型參數(shù)

根據(jù)表1 中的參數(shù)，下面逐個(gè)計(jì)算每個(gè)過程的開銷，最終得到D-CACHEJOIN 算法的整體吞吐率μ（單位為元組/秒）。

1) 數(shù)據(jù)在網(wǎng)絡(luò)中傳輸?shù)拈_銷。假設(shè)分布式系統(tǒng)由n個(gè)節(jié)點(diǎn)構(gòu)成，單位時(shí)間內(nèi)需轉(zhuǎn)發(fā)的節(jié)點(diǎn)數(shù)為，因此網(wǎng)絡(luò)中數(shù)據(jù)總的傳輸開銷為。

3) 讀取關(guān)系表R中k個(gè)分頁的開銷為CI/O(k)。

4) 移除w個(gè)匹配完成的元組及其指針的開銷為wCE。

5) 從流緩沖區(qū)中讀取w個(gè)元組的開銷為wCS。

6) 將w個(gè)元組加入HS的開銷為wCA。

7) 將HS中元組與R中元組匹配的開銷為。

8) 輸出結(jié)果的開銷。結(jié)果輸出時(shí)的開銷和連接的成功率有關(guān)，成功率越高，需要輸出的結(jié)果元組就越多，因此，最終輸出結(jié)果的開銷為。

由此可得最終D-CACHEJOIN 算法的整體吞吐率為

用nSP表示流探測階段中每次迭代處理的元組數(shù)量，nDP表示磁盤探測階段中每次迭代處理的元組數(shù)量，則算法的吞吐率還可以表示為

式(1)和式(2)能否精確表示流數(shù)據(jù)和關(guān)系表進(jìn)行連接操作的開銷的前提是算法必須只需要有限的內(nèi)存就可以持續(xù)運(yùn)行，同時(shí)流數(shù)據(jù)S中新到達(dá)的元組進(jìn)入系統(tǒng)時(shí)，需要有剩余的緩沖區(qū)內(nèi)存對其進(jìn)行保存。更加詳細(xì)的關(guān)于關(guān)系表R的分頁數(shù)和流緩沖區(qū)中元組個(gè)數(shù)的理論論證可參考文獻(xiàn)[6]。

2.2 算法成本模型

由式(2)可知，算法的吞吐率與R的設(shè)置情況、S的設(shè)置情況、流和磁盤緩沖區(qū)等因素有關(guān)，本節(jié)通過定量計(jì)算來描述算法的成本。

1)nSP的模型計(jì)算

在流檢測階段中，S加載到流緩沖區(qū)后立即與HR中的元組進(jìn)行匹配，且優(yōu)先與HR中較多次與R中元組成功匹配的元組進(jìn)行，如果匹配成功，則直接生成結(jié)果輸出；如果匹配失敗，則該元組在后續(xù)的磁盤檢測階段嘗試進(jìn)行匹配。由此，nSP與HR、R中元組大小、R中元組個(gè)數(shù)、流緩沖區(qū)大小有關(guān)，其中流緩沖區(qū)大小在后續(xù)實(shí)驗(yàn)中得知極?。ㄊ冀K在0.1 MB 以下），因此忽略不計(jì)。

根據(jù)表1，nR為R中元組個(gè)數(shù)，為HR中元組個(gè)數(shù)。現(xiàn)實(shí)中在構(gòu)建R和S時(shí)，由于其數(shù)據(jù)分布并不均勻，需要考慮其扭曲分布系數(shù)Zipf，如果Zipf 值為0，那么連接屬性值完全均勻分布，隨著Zipf 值的增大，數(shù)據(jù)的不均勻分布程度就越大。假設(shè)流數(shù)據(jù)S有S～Zipfian(1,n)，其概率密度函數(shù)為，累積分布函數(shù)為，其中，Hx,1同理[20]。通過歸一化處理，流檢測階段的匹配概率p1可表示為

其中，x和y是未知數(shù)。

當(dāng)nhR變?yōu)樵瓉淼? 倍時(shí)，式(4)變?yōu)?/p>

將式(5)代入式(4)，得φ1=2x，即x=lbφ1。

當(dāng)nR變?yōu)樵瓉淼? 倍時(shí)，式(4)變?yōu)?/p>

將式(6)代入式(4)，得ψ1=2y，即y=lbψ1。

因此，式(4)變?yōu)?/p>

若在n次迭代中算法處理的流元組總量為Sn，則有

2)nDP的模型計(jì)算

在流檢測階段，算法已經(jīng)通過HR連接匹配了S中的部分元組，在磁盤檢測階段中，將使用常規(guī)方式進(jìn)行關(guān)系表R和流數(shù)據(jù)S中元組的連接。由于R較大，算法分段對R進(jìn)行讀取，每次讀取磁盤緩沖區(qū)大小nRB的數(shù)據(jù)段，由于S～Zipfian(1,n)，因此可以逐段計(jì)算匹配概率后相加，表示為

記R中元組的分段數(shù)為N，由匹配概率之和經(jīng)歸一化后可得平均匹配概率p2為

類似于nSP的計(jì)算方式，p2可以用nRB、nR和nhR表示?，F(xiàn)考察nRB、nhR和nR分別變?yōu)? 倍時(shí)p2的變化情況。假設(shè)此時(shí)p2分別增大一個(gè)常數(shù)因子θ2、增大一個(gè)常數(shù)因子φ2和減小一個(gè)常數(shù)因子ψ2，且有

和nSP中處理方法相同，此時(shí)可得x=lbφ2，y=lbψ2，z=lbθ2。

因此，式(11)可以變?yōu)?/p>

若哈希表中存儲(chǔ)的流元組總量為，則有

確定nSP和nDP后，即可通過式(2)對算法進(jìn)行調(diào)整。

2.3 實(shí)驗(yàn)設(shè)計(jì)

實(shí)驗(yàn)環(huán)境為4 臺(tái)IBM POWER x-236 8841 服務(wù)器，Intel Xeon Gold 6248R CPU，64 GB 內(nèi)存，Linux 64 位操作系統(tǒng)，具體配置情況如表2 所示。

表2 實(shí)驗(yàn)環(huán)境

環(huán)境部署的規(guī)模主要從數(shù)據(jù)訪問峰值、吞吐量要求和響應(yīng)時(shí)間等綜合考量?；舅悸肥亲x寫分離。通過解析查詢語句，對于僅包括讀操作的配置一個(gè)連接字符串到讀服務(wù)器，對于寫操作則配置另一個(gè)連接字符串到寫服務(wù)器，將大規(guī)模數(shù)據(jù)的訪問分流到多臺(tái)服務(wù)器上，使應(yīng)用中讀取數(shù)據(jù)的速度和并發(fā)量顯著提高，增加了系統(tǒng)響應(yīng)速度，減少了服務(wù)器的壓力，能夠有效增加系統(tǒng)的穩(wěn)定性和擴(kuò)展性。在讀寫分離的基礎(chǔ)上再進(jìn)行負(fù)載均衡，具體過程是各節(jié)點(diǎn)通過專用網(wǎng)絡(luò)進(jìn)行連接，對每個(gè)服務(wù)器進(jìn)行監(jiān)測，獲取資源占用情況，整個(gè)集群可以視為一臺(tái)具有超高性能的獨(dú)立服務(wù)器。

實(shí)驗(yàn)數(shù)據(jù)來自某運(yùn)營商內(nèi)部的原始話單數(shù)據(jù)，以此構(gòu)建流數(shù)據(jù)S和關(guān)系表R，各包含約2 000 萬條數(shù)據(jù)。

由于本文算法具有緩存模塊和自動(dòng)控制緩存中元組換入換出的閾值的功能，因此需要一個(gè)熱身階段。由于在實(shí)際場景中，程序開始后會(huì)持續(xù)地運(yùn)行，因此熱身階段是可以接受的，后續(xù)實(shí)驗(yàn)中如果沒有特殊說明，該系統(tǒng)熱身階段都會(huì)忽略。

本節(jié)設(shè)計(jì)實(shí)驗(yàn)測試了本文算法的性能，實(shí)驗(yàn)的具體軟硬件條件已在表2 中列出，在此環(huán)境上搭建了具有8 個(gè)虛擬節(jié)點(diǎn)的分布式集群。通過將本文算法與傳統(tǒng)的MESHJOIN 算法、CACHEJOIN 算法和較先進(jìn)的Eunomia 分布式連接方法進(jìn)行比較，對算法的執(zhí)行效率進(jìn)行實(shí)驗(yàn)考察，各實(shí)驗(yàn)運(yùn)行5 次后取平均值作為實(shí)驗(yàn)結(jié)果。

2.4 實(shí)驗(yàn)結(jié)果及分析

實(shí)驗(yàn)1不同數(shù)據(jù)量對算法效率的影響

首先令集群開啟4 個(gè)節(jié)點(diǎn)，處理不同數(shù)量級的數(shù)據(jù)，吞吐率對比如圖4 所示，D-CACHEJOIN都取得了比其他3 種算法更好的吞吐率，這一方面是因?yàn)閷?shí)驗(yàn)數(shù)據(jù)并不是均勻分布，而是服從Zipfian 分布的，其中一部分?jǐn)?shù)據(jù)具有比其他數(shù)據(jù)更多的出現(xiàn)次數(shù)，沒有對此進(jìn)行額外考慮的算法無法適應(yīng)該環(huán)境；另一方面是因?yàn)镈-CACHEJOIN具有緩存模塊，能夠較好地處理數(shù)據(jù)傾斜的情況。同時(shí)，D-CACHEJOIN 的吞吐率在10 萬、100 萬和1 000 萬條數(shù)據(jù)的情況下分別是Eunomia的1.30 倍、1.48 倍和1.65 倍，算法的優(yōu)勢隨著數(shù)據(jù)量的增多而逐漸增大，這是因?yàn)閿?shù)據(jù)量增多時(shí)，可能對磁盤的 I/O 次數(shù)逐漸增大，D-CACHEJOIN 越來越多地減少了實(shí)際上對于磁盤的I/O 次數(shù)。

圖4 不同數(shù)據(jù)量下算法的吞吐率

現(xiàn)實(shí)中大多數(shù)的大型數(shù)據(jù)集是服從Zipfian 分布的[21]，包括本文實(shí)驗(yàn)的數(shù)據(jù)集。相對于均勻分布的數(shù)據(jù)集，對服從Zipfian 分布的數(shù)據(jù)集進(jìn)行實(shí)驗(yàn)有更大的實(shí)際意義，但為了實(shí)驗(yàn)的完整性，給出其他實(shí)驗(yàn)條件相同時(shí)對均勻數(shù)據(jù)的實(shí)驗(yàn)結(jié)果，如圖5 所示。

圖5 不同數(shù)據(jù)量下算法的吞吐率（均勻分布）

實(shí)驗(yàn)2可用內(nèi)存總量對算法效率的影響

本文實(shí)驗(yàn)中可用內(nèi)存的總量為磁盤緩沖區(qū)、流緩沖區(qū)、隊(duì)列、HR和HS的大小之和，但本文實(shí)驗(yàn)中有兩點(diǎn)不變：流緩沖區(qū)基本不變而且可以忽略，因?yàn)樵谒星闆r下流緩沖區(qū)使用的內(nèi)存不超過0.1 MB；H R保持不變，因?yàn)镠R的大小會(huì)在很大程度上影響本文算法的效率，所以在后續(xù)實(shí)驗(yàn)中專門對HR的大小對算法效率的影響進(jìn)行研究，故在本文實(shí)驗(yàn)中HR保持為R的1%，一旦HR滿，如果再次有元組的匹配頻率超過threshold Value，將可能會(huì)對HR中已有元組進(jìn)行替換并可能引起threshold Value 的動(dòng)態(tài)調(diào)整?？捎脙?nèi)存總量對算法效率的影響如圖6 所示。

圖6 可用內(nèi)存總量對算法效率的影響

從圖6 可以看出，幾種流連接算法的吞吐率都會(huì)隨著可用內(nèi)存總量的增加而提高，由于引入了緩存模塊，CACHEJOIN 和D-CACHEJOIN 的吞吐率都要比MESHJOIN 高2 倍以上，而Eunomia 由于需要基于時(shí)間戳對流數(shù)據(jù)進(jìn)行檢查，因此需要較多的額外內(nèi)存才能表現(xiàn)出較高的效率，內(nèi)存較少時(shí)執(zhí)行效率不足。另有其他實(shí)驗(yàn)顯示，當(dāng)提供大量內(nèi)存時(shí)，Eunomia 的吞吐率已經(jīng)超過了CACHEJOIN，但仍低于D-CACHEJOIN。考慮到實(shí)驗(yàn)的完整性，補(bǔ)充了內(nèi)存較少時(shí)的情況，由于D-CACHEJOIN 需要存儲(chǔ)哈希函數(shù)的計(jì)算結(jié)果以便后續(xù)使用，因此當(dāng)可用內(nèi)存較少時(shí)，D-CACHEJOIN 的吞吐率要略低于CACHEJOIN。由于本文研究的支撐系統(tǒng)具有充足的內(nèi)存可供使用，同時(shí)鑒于現(xiàn)代分布式系統(tǒng)中如果有計(jì)算需要，通常都會(huì)配備充足的內(nèi)存，因此一般不會(huì)出現(xiàn)內(nèi)存嚴(yán)重不足的情況。當(dāng)可用內(nèi)存增大后，D-CACHEJOIN 的吞吐率逐漸超過CACHEJOIN，且隨著內(nèi)存的增加，前者吞吐率的增加速度要高于后者；當(dāng)內(nèi)存達(dá)到100 MB 時(shí)，D-CACHEJOIN 的吞吐率相比于CACHEJOIN 增加了10%以上。

實(shí)驗(yàn)3R的緩存比例對算法效率的影響

由于算法引入了緩存模塊，因此可對R中頻繁匹配的元組進(jìn)行緩存，能夠大大提升算法的效率，本文實(shí)驗(yàn)固定可用內(nèi)存大小，研究R的緩存比例對算法效率的影響，實(shí)驗(yàn)結(jié)果如圖7 所示。

圖7 R 的緩存比例對算法效率的影響

根據(jù)圖7 可以看出，當(dāng)R的緩存比例達(dá)到10%時(shí)，D-CACHEJOIN 算法的吞吐率是MESHJOIN 算法的10 倍以上。當(dāng)緩存比例較低時(shí)，D-CACHEJOIN算法的吞吐率仍然是MESHJOIN 算法的2 倍以上，由于沒有引入緩存策略，MESHJOIN 和Eunomia的吞吐率保持不變。這說明引入緩存模塊可以大大提升算法的性能。

實(shí)驗(yàn)4D-CACHEJOIN 算法在分布式集群中的擴(kuò)展性

為了測試算法在分布式集群中的擴(kuò)展情況，將集群中的節(jié)點(diǎn)數(shù)n由4 增加到8，并且改變算法處理的數(shù)據(jù)量，考察此時(shí)的執(zhí)行效率，實(shí)驗(yàn)結(jié)果如圖8 所示。

圖8 算法吞吐率隨節(jié)點(diǎn)擴(kuò)展的變化情況

總體而言，各算法的吞吐率隨著節(jié)點(diǎn)數(shù)的增多而增大，且D-CACHEJOIN 保持著最佳的執(zhí)行效率。圖8 中從左至右，D-CACHEJOIN 的吞吐率依次是Eunomia 的1.29 倍、1.15 倍、1.48 倍、1.36 倍、1.65 倍、1.50 倍，這表示隨著集群中節(jié)點(diǎn)數(shù)的增加，D-CACHEJOIN 相對于Eunomia 在執(zhí)行效率上的優(yōu)勢變小了（對其他2 種算法也有這一結(jié)論）。這是因?yàn)殡S著節(jié)點(diǎn)數(shù)的增多，每個(gè)節(jié)點(diǎn)上關(guān)系表R的片段就越小，用于磁盤 I/O 的時(shí)間就越短，而D-CACHEJOIN 恰恰減少了磁盤I/O 的次數(shù)，因此這一優(yōu)勢在節(jié)點(diǎn)數(shù)增加時(shí)變小了。但與對內(nèi)存的訪問相比，磁盤I/O 是非常消耗時(shí)間的操作，因此即使集群進(jìn)一步擴(kuò)容，只要磁盤上的關(guān)系表無法全部放入內(nèi)存中，D-CACHEJOIN 算法仍然會(huì)有更優(yōu)的執(zhí)行效率。

綜上，D-CACHEJOIN 算法總體上保持了最優(yōu)的執(zhí)行效率。當(dāng)ETL 系統(tǒng)中數(shù)據(jù)的數(shù)量級發(fā)生比較大的變化時(shí)，由于對數(shù)據(jù)庫的訪問時(shí)間，即對磁盤的I/O 所需時(shí)間會(huì)隨之增大，算法的吞吐率會(huì)隨之下降，但本文算法的優(yōu)勢也會(huì)逐漸增大，因?yàn)楸疚乃惴p少了對磁盤的I/O 次數(shù)。為了擁有更高的效率，算法往往會(huì)進(jìn)行更多消耗內(nèi)存的額外工作，在現(xiàn)代處理海量數(shù)據(jù)的分布式系統(tǒng)中，通常都會(huì)配備足夠的內(nèi)存，同時(shí)本文算法擁有一定的自由度，可以根據(jù)實(shí)際情況設(shè)置緩存的大小，更好地貼合具體場景，本文算法一般能夠很好地適應(yīng)環(huán)境并良好運(yùn)行。當(dāng)分布式集群中新增或下線節(jié)點(diǎn)時(shí)，本文算法保持了更優(yōu)的執(zhí)行效率；當(dāng)處理海量數(shù)據(jù)時(shí)，本文算法在分布式集群中具有良好的拓展性。

3 結(jié)束語

本文介紹了流數(shù)據(jù)接收過程中面臨的主要問題，并簡單介紹了已有的流連接算法的處理策略，在此基礎(chǔ)上對算法進(jìn)行改進(jìn)，提出了一種把使用緩存的流連接策略應(yīng)用于分布式環(huán)境的D-CACHEJOIN算法。此外，本文詳細(xì)描述了該算法的執(zhí)行架構(gòu)并計(jì)算其性能開銷，通過實(shí)驗(yàn)展示了該算法的執(zhí)行效率，說明了該算法具有較好的適應(yīng)性，在一般的大數(shù)據(jù)處理系統(tǒng)中能夠良好運(yùn)行，并且在分布式集群中具有良好的拓展性。