潘兆平　張建軍　魏志強

摘要：本文介紹了分布式流數(shù)據(jù)的在線匯聚合與統(tǒng)計的方法，該方法采用在分布式隨機采樣算法的基礎上增加了一個權重的概念，它可以從分布式流數(shù)據(jù)中進行隨機采樣。該方法把多個在線查詢任務分解成一個多層次處理單元集合，每個處理單元負責一個時段的數(shù)據(jù)查詢，這些處理單元能夠并行處理，在并行處理過程中，流數(shù)據(jù)以事件方式封裝打包，通過處理單元之間的相互配合完成整個查詢任務。在多層次查詢過程中，處理單元能將一些重復性的計算進行合并，這樣就避免重復計算帶來的消耗，提高查詢語句的執(zhí)行效率。

關鍵詞：匯聚與統(tǒng)計；分布式；流數(shù)據(jù)；隨機采樣

中圖分類號：TP311.13 文獻標識碼：A 文章編號：1007-9416（2018）09-0140-04

1 引言

過去，用戶從一個大型數(shù)據(jù)庫在線查詢數(shù)據(jù)時，需等待系統(tǒng)遍歷完所有數(shù)據(jù)后才能得到最終查詢結果，處理過程中沒有任何反饋信息。

在查詢過程中，為了實時地獲得數(shù)據(jù)查詢的中間狀態(tài)，可以采取降低數(shù)據(jù)的準確性的策略來實現(xiàn)查詢信息的及時反饋。在線匯聚與統(tǒng)計方法就是連續(xù)為用戶提供近似查詢結果，同時輸出相應的統(tǒng)計結果，而不是等待處理完所有數(shù)據(jù)后再給出最終結果；隨著在線處理的數(shù)據(jù)越來越多，匯聚與統(tǒng)計結果持續(xù)更新，直到結果能夠滿足用戶需求為止。

2 實現(xiàn)方法

在線匯聚與統(tǒng)計能連續(xù)地提供實時匯聚與統(tǒng)計分析結果，本文采用的流數(shù)據(jù)處理方法具有在線匯聚能力，當在線接收到一組數(shù)據(jù)時，系統(tǒng)能快速地處理這組數(shù)據(jù)，并將匯聚與統(tǒng)計結果以異步的方式發(fā)送給用戶。文獻[1]提出，流數(shù)據(jù)操作有兩種：無狀態(tài)操作和有狀態(tài)操作。無狀態(tài)操作（如映射、并集等）根據(jù)輸入數(shù)據(jù)進行處理，而后把結果發(fā)送給下一個操作即可，是不需要保留中間計算結果。而有狀態(tài)操作（如合并，笛卡爾乘積）一般采用滑動窗口（可以是固定時間窗口，也可以是固定個數(shù)窗口）方法進行處理，需要一串數(shù)據(jù)同時操作。

為了支持狀態(tài)操作，本文采用滑動窗口方式進行在線匯聚，通過時間窗口與順序窗口兩個條件同時觸發(fā)狀態(tài)匯聚操作。當在線接收到流數(shù)據(jù)時，系統(tǒng)將連續(xù)處理，并輸出中間結果。如圖1，系統(tǒng)包含采樣、映射、匯聚 和統(tǒng)計等處理過程。

2.1 分布式流數(shù)據(jù)的隨機采樣

文獻[2]指出，蓄水池采樣算法可以從N個流數(shù)據(jù)中隨機采樣，其中N是一個很大的數(shù)，這些數(shù)通常不保存在內存中。在初始狀態(tài)下，該算法有s個采樣數(shù)據(jù)，當?shù)趇個數(shù)據(jù)到達時（其中i>s），有s/i的概率選擇這個新數(shù)據(jù)。具體做法是：從數(shù)據(jù)[1，i]中隨機得到一個值j，如果j≤s，把第i個數(shù)據(jù)替換掉內存中的第j個數(shù)據(jù)，否則丟棄這個數(shù)（第i個數(shù)）。這個算法只需要O（s）個空間和O（1）的采樣時間。

但是，蓄水池采樣算法無法從分布式流數(shù)據(jù)中進行隨機采樣，原因有兩個：第一個是不同流數(shù)據(jù)可能采用不同的數(shù)據(jù)分布（文獻[3]），如一個采用伯努利分布，另一個采用吉布斯。第二個是來自不同流數(shù)據(jù)有不同的權重，如網絡監(jiān)控和網頁服務等。

為了解決上述問題，分布式流數(shù)據(jù)采樣算法引入了一個權重的概念，該算法不僅能從不同的流數(shù)據(jù)中進行隨機采樣，而且這些流數(shù)據(jù)還帶有不同的權重。分布式流數(shù)據(jù)采樣算法拓撲結構見圖2，圖中有m個數(shù)據(jù)流，有m個本地采集器，每個數(shù)據(jù)流含有n個數(shù)據(jù)。本地采集器Pi對流數(shù)據(jù)Si進行采樣，并把采樣結果發(fā)送給M（M可以是一個有狀態(tài)操作，也可以是一個無狀態(tài)操作），每個采樣結果都是帶有權重的。

圖2中，一個流數(shù)據(jù)端Si將以Pi的概率選擇接收到的數(shù)據(jù)e（該數(shù)權重為wi），并把數(shù)據(jù)e發(fā)送給M。M中維護了一個數(shù)據(jù)候選集V（匯集了多個流數(shù)據(jù)的采樣結果）。對于任何一個流數(shù)據(jù)vi（vi∈V），分配一個鍵值ki（ki=ui1/wi，ui是一個從0到1范圍內隨機的一個值），M總是從V中選擇s個數(shù)據(jù)（其鍵值ki最大）作為最終采樣數(shù)據(jù)。由于ui是個隨機數(shù)，因此，ki就是個隨機值，在加上s個數(shù)據(jù)是隨機選擇的，因此這s個數(shù)據(jù)是來自Si的隨機采樣結果。

2.2 分類匯聚處理方法

在線匯聚系統(tǒng)框圖中，“映射器”負責對數(shù)據(jù)進行分類，將數(shù)據(jù)分配到不同的匯聚器上。匯聚器對數(shù)據(jù)進行過濾與篩選，而后將某個特定時間段的數(shù)據(jù)進行“合并”操作，最后把結果數(shù)據(jù)發(fā)送給統(tǒng)計器作統(tǒng)計；通過統(tǒng)計計算得出近似查詢結果。

2.2.1 數(shù)據(jù)映射

由于來自不同數(shù)據(jù)流的數(shù)據(jù)格式可能不一樣，而且存在無效值，因此，在數(shù)據(jù)進映射前，需要對采樣到的流數(shù)據(jù)進行清洗（即統(tǒng)一格式），將每個數(shù)據(jù)轉化成的數(shù)據(jù)塊，其中key是數(shù)據(jù)的鍵值，value是數(shù)據(jù)的屬性值。數(shù)據(jù)經過清洗后發(fā)送給映射器，由于不同數(shù)據(jù)塊含有不同的key，因此“映射器”通過key對數(shù)據(jù)塊分類，并把分類結果發(fā)送給相應的匯聚處理單元進行處理，匯聚處理單元之間可并發(fā)執(zhí)行，同時單個機器節(jié)點可以分配多個匯聚處理單元。

2.2.2 流數(shù)據(jù)匯聚

在匯聚器中有許多處理單元（簡稱：PE），這些PE間相互獨立，可以并行處理數(shù)據(jù)；每個PE中的數(shù)據(jù)具有相同的“key”，PE對收到的數(shù)據(jù)進行過濾與篩選，其方法與數(shù)據(jù)庫中的行過濾與列過濾相同，這樣可以減少匯聚的計算成本與數(shù)據(jù)通信成本。SQL查詢語句參見圖3。

分布式流數(shù)據(jù)匯聚操作見圖4，每個PE中有兩個列表（流A和流B），它們之間通過key值進行匯聚操作得到相關結果。給定一個key，PE將從表流A和流B中檢索相關數(shù)據(jù)，并對相關數(shù)據(jù)進行匹配與連接。流A和B的數(shù)據(jù)經過采樣后存儲在內存中，系統(tǒng)在一個時間窗口內自動執(zhí)行合并任務，當一個時間窗口的合并操作結束后（當前窗口的數(shù)據(jù)從內存中釋放掉，并等待下一個時間窗口的運行），將通知匯聚開始這個窗口的數(shù)據(jù)匯聚。

在實際中，流數(shù)據(jù)往往存在不同步和無序的特性，這樣的數(shù)據(jù)流可能導致時間窗口不完整，因此，檢測一個時間窗口是否結束就變得十分復雜。在這里將采用標記方式來指示一個時間窗口合并操作是否結束，當匯聚操作單元接收到所有標記通知時，才開始這個窗口的匯聚操作和更新狀態(tài)。

由于內存有限，本方案只把部分PE常駐于內存中，如果某個PE已經完成任務，則移除對應的PE，釋放出相應的內存，但是建立與移除PE時耗費了一定的時間和計算能力，因此，為了提高機器的計算能力、減少不必要的消耗，本文采用計算單元重用的方法：當內存不足而有其它PE空閑時，系統(tǒng)將自動使用這些空閑PE；當內存足夠而所有的PE都在忙時，那么將為該任務重新創(chuàng)建一個PE。

3 統(tǒng)計分析

匯聚操作結束后，系統(tǒng)對匯聚結果進行統(tǒng)計評估，計算其置信區(qū)間與誤差范圍；根據(jù)單表匯聚操作與多表匯聚操作的不同，其計算置信區(qū)間與誤差范圍時的方法也有一些區(qū)別。

3.1 單表匯聚查詢

單表匯聚查詢形式可以簡單表示如下：

SELECT opt（Exp（xi）） FROM A.a

流數(shù)據(jù)匯聚操作采用了一個基于時間窗口的策略：若在第i個時間窗口（τ=Ti-Ti-1）PE存有m個樣本數(shù)據(jù)（x1，x2，...，xm），那么，匯聚時該PE需要計算四個值（Ni是第i個時間窗口的數(shù)據(jù)個數(shù)，是第i個時間窗口中所有數(shù)據(jù)的平均數(shù)，Sumi是第i個窗口所有數(shù)據(jù)的總和，σi2是第i個時間窗口中數(shù)據(jù)的方差值），求解公式如下：

根據(jù)公式3-4，如果給定一個置信度p，可以計算出∈的值，得到誤差范圍為[λ-∈，λ+∈]；或者給定一個誤差范圍∈，也可以同樣計算出這個范圍內置信度的值p。

3.2 多表匯聚查詢

多表匯聚查詢形式可以表示如下：

SELECT opt（Exp（xi，xj））FROM A，B

WHERE A.a=B.b

要想計算多表匯聚查詢的置信區(qū)間和誤差范圍，需要對每個表的方差值進行分別計算，根據(jù)文獻[4]中的表述，多表匯聚查詢結果的計算公式如下：

其中σA和σB是表A和B的近似方差值，因此，一個基于時間窗口的匯聚結果可以輸出為：（Ti，Ti-1，Ni，，Sumi，σi2），奪標匯聚查詢結果與單表匯聚查詢結果相似。

4 實驗結果評估

4.1 實驗環(huán)境配置

分布式流數(shù)據(jù)的在線匯聚與統(tǒng)計算法部署在24臺機器中，每臺機器的硬件配置：CPU因特爾E2600，內存16GB，硬盤1TB；每臺機器的軟件配置：Apache S4 系統(tǒng)平臺，編程語言是JAVA；機器之間采用萬兆網絡。在實驗過程中，將1臺機器配置管理節(jié)點，另外20臺機器作為計算節(jié)點，剩余3臺機器作為備用機器（故障檢查及容錯）。

實驗數(shù)據(jù)有四張表，總數(shù)據(jù)量約10GB，其中Litem表含有1300萬行記錄，Orders表有350萬行記錄， Part表有80萬行記錄，表Psup有160萬行記錄。

本次實驗采用以下三個SQL語句來評估分布式流數(shù)據(jù)在線匯聚與統(tǒng)計算法的實際性能，包含匯聚、統(tǒng)計和多語句同時查詢。實驗的結果都是經過多次運行后計算所得的平均值，整個SQL的運行完成時間是從提交SQL查詢到執(zhí)行完畢為止。

查詢模版1（Q1）：

SELECT Sum（Ext*Dis），Average（Ext*Dis）

FROM Litem

WHERE Dis<0.15 and Dis>=0.07

查詢模版2（Q2）：

SELECT A.ReturnFlag， A.LineStatus，B.OrderPriority，

Sum（A.Quantity），Count（A.*）

FROM Litem A， Orders B

WHERE A.OrderKey=B.OrderKey

and A.Dis<0.15 and A.Dis>=0.07

AND B.TotalPrice>=1000

AND B.TotalPrice<30000

GROUP BY A.RFlag，A.LStatus，B.OPriority

查詢模版3（Q3）：

（1）SELECT B.MFGR，B.BRAND，

Sum（A.Quantity），

Average（A.Quantity）

FROM Litem A，Part B

WHERE A.PartKey = B.PartKey

AND A.Dis<0.15 AND A.Dis>=0.07

GROUP BY B.MFGR，B.BRAND

（2）SELECT A.ReturnFlag， A.LStatus，Sum（B.SupplyCost），

Average（B.SupplyCost）

FROM Litem A，Psup B

WHERE A.PartKey = B.PartKey

AND A.Dis<0.15 AND A.Dis>=0.07

GROUP BY A.RFlag， A.LStatus

在對數(shù)據(jù)進行匯聚與統(tǒng)計計算時，時間窗口中緩存的數(shù)據(jù)量大小直接影響在線匯聚與統(tǒng)計算法的運行時間，因此首先評估了數(shù)據(jù)量大小與SQL的平均運行時間關系，然后評估了置信區(qū)間與誤差區(qū)間之間的影響，分析數(shù)據(jù)量大小與執(zhí)行時間的關系；通過調整集群中機器的個數(shù)（從2到22）進行比較分析，最后比較Spark與S4在處理流數(shù)據(jù)中性能。

4.2 多查詢語句性能分析

通過實驗分析多查詢語句并行工作的性能，試驗中實現(xiàn)的查詢語句為兩個Q3，這兩個查詢包含相同的濾波條件，濾波操作與結果可以合并成一個；但在匯聚操作中關聯(lián)的表是不同的，因此，需要把濾波的數(shù)據(jù)結果分兩塊發(fā)送給各自的匯聚節(jié)點，通過相同操作合并機制，可以減少不同查詢語句的重疊計算。

通過調整數(shù)據(jù)的大?。◤?G到10G）進行測試，多查詢語句Q3與子查詢Q3-1和Q3-2分別運行在同一集群和同一數(shù)據(jù)集中。進行多次試驗后平均得到結果，如圖5顯示。

當數(shù)據(jù)量增加時，所有查詢的運行時間也隨之增加，獨立并行執(zhí)行完查詢Q3-1和Q3-2的時間比共享的運行時間多。我們計算了通過該拓撲結構帶來的性能提高，假設Q3的運行時間是t3，Q3-1與Q3-2的運行時間分別是t31和t32，那么合并的查詢Q3的提高為（t31+t32-t3）÷（t31+t32）。圖6描述了平均提高效果為18%。

5 結語

在本文中，在分布式流數(shù)據(jù)系統(tǒng)S4中實現(xiàn)了一個在線匯聚的方法，采用“行為模式”簡化復雜查詢，支持流水線任務和并行處理；著重分析了MapReduce Online與本章方法的區(qū)別，并比較了兩者的結果。由于MapReduce Online在處理流水線任務時，會把中間結果存儲在硬盤中，而執(zhí)行結果直接從內存中發(fā)送給下一個任務，實驗結果顯示該方法有效地支持實時數(shù)據(jù)匯聚；在處理多查詢方面，通過一個拓撲結構指導查詢的執(zhí)行，該拓撲結構能夠合并數(shù)據(jù)的重復操作，減少了系統(tǒng)計算工作量，實驗結果顯示拓撲結構能夠很大程度上提高整體的查詢效率；通過公開的基準數(shù)據(jù)TPC-H進行實驗，結果顯示該方法能夠把較準確的結果快速反饋給用戶，運行的速度與效果也都比MapReduceOnline好。

參考文獻

[1]ABADI D， CARNEY D， C， ETINTEMEL U， et al. Aurora： a new model and architecture fordata stream management[J]. The international Journal on VLDB，2003，12（2）：120-139.

[2]VITTER J S. Random sampling with a reservoir[J]. ACM Transactions on Mathematical Software，1985，11（1）：37-57.

[3]CORMODE G， MUTHUKRISHNAN S， YI K， et al. Optimal sampling from distributed streams[C].Principles of Database Systems.[S.l.]：ACM，2010：77-86.

[4]HAAS P， HELLERSTEIN J. Ripple joins for online aggregation[C].ACM SIGMOD international conference on Management of data.[S.l.]：ACM，1999，28：287-298.