杜政頡, 王 鵬, 黃 焱, 郎福通

(1.成都信息工程學院軟件工程學院并行計算實驗室,四川成都610225;2.中國科學院成都計算機應用研究所,四川 成都 610041;3.中國科學院大學,北京 100049)

0 引言

大數(shù)據(jù)處理分為兩類模式,一類是批處理模式,數(shù)據(jù)源為靜態(tài);一類是流處理模式,數(shù)據(jù)源為動態(tài)。批處理模式系統(tǒng)有Hadoop、Spark、Disco、HPCC等,流處理模式系統(tǒng)來自 Twitter的 Storm和來自 Yahoo的 S4系統(tǒng)[1]。MapReduce是Hadoop中被廣泛應用的一種大數(shù)據(jù)處理模型,側重于批處理,Topology是來自于Storm系統(tǒng)中的一種編程模型,側重于流處理。

MapReduce無法解決具有迭代結構的應用程序,迭代結構程序在實際應用中很普遍,因此,有人基于MapReduce提出一種迭代MapReduce。文獻[2]提出一種名為Twister的迭代MapReduce處理方案,文獻[3]提出一種名為HaLoop的MapReduce迭代方案。

Storm是一款應用于實時流處理領域的大數(shù)據(jù)處理工具。在Storm中,Nathan Mar提出一種新的并行編程模型Topology。這種模型改進了MapReduce需要存儲中間數(shù)據(jù)這一繁瑣過程[2],采用類似于流水線作業(yè)方式的任務分解模型,側重于處理動態(tài)數(shù)據(jù)源的任務,實時性更強。Storm編程模型與MapReduce一樣,并沒有考慮這種迭代結構應用程序的實現(xiàn)過程。對于這一缺陷,還沒有人提出一種改進方案。因此,文中基于這種Topology編程模型,通過增加組件Receiver、IBolt、Checker組建迭代Topology,設計了一種新的可以解決迭代結構應用程序Topology模型,并對這種模型的新增組件和其對應的API進行了介紹和分析,在Storm系統(tǒng)架構基礎上設計了一種迭代Topology的實現(xiàn)方案,描述了在這種實現(xiàn)方式下解決具有迭代結構程序的具體過程,并使用這種模型實現(xiàn)了K-Means算法,實例論證這種迭代模型的可行性。

1 問題描述

1.1 Storm編程模型思想

Storm編程模型原理[4]:一般任務都可以用流水線作業(yè)方式表現(xiàn)出來,其中的組件就相當于流水作業(yè)中的一個工人,不同的組件負責任務中不同的部分,一個組件處理完自己的工作即提交給下一個組件,直至整個任務處理完成。整個任務實現(xiàn)的過程可以用圖1的Topology來表示。這種模型的一個突出特點:數(shù)據(jù)源可以靜態(tài)可以動態(tài),動態(tài)環(huán)境中它的表現(xiàn)更能體現(xiàn)出它的優(yōu)勢。一方面,這種模型采用消息傳遞方式交互數(shù)據(jù),數(shù)據(jù)量相比于從磁盤獲取要小,動態(tài)環(huán)境中,數(shù)據(jù)量動態(tài)讀取,每次讀取量小,很好滿足了這種模型的特點;另一方面,這種模型是一種實時性處理模型,動態(tài)環(huán)境中更能夠體現(xiàn)這種特點。所以,這種處理模型側重于流處理。模型包含兩類組件:Spout和Bolt,Spout組件負責讀取數(shù)據(jù)源,Bolt組件負責實際的數(shù)據(jù)操作運算。組件在實現(xiàn)中有對應的API,Spout組件API為setSpout(),Bolt組件API為setBolt()。

圖1 Storm Topology

1.2 Storm編程模型解決迭代時的不足

有許多并行算法內部都帶有簡單的迭代結構。這些算法大多分布在數(shù)據(jù)聚類、維度縮減、鏈接分析、機器學習和計算機視覺等領域。K-Means、確定性退火聚類、PageRank和SMACOF算法就是其中的例子。這種具有迭代結構的算法可以用以下公式來描述[3]:

其中,R0表示初始化時的結果,L表示一種不變的關系。這種公式表示的程序,只有當?shù)竭_某檢查點時才將終止運行。比如,迭代后的結果與前面結果相比已經不會出現(xiàn)變化就可以作為一個檢查點,這也是很多優(yōu)化算法迭代的檢查點。

迭代程序的一個關鍵點是結果與輸入具有相關性,上面的Topology中輸出結果無法再次作為輸入,所以這種迭代部分只能放入一個組件Bolt作為一個任務來處理,這樣就增加了該組件的處理負載。對于低配置集群,這種高負載一方面導致整體系統(tǒng)效率降低,另一方面還有可能使節(jié)點失效。因此,下面設計一種具有迭代結構的Topology模型,在這種模型中,處理迭代結構程序時,就可以把迭代結構部分按照功能拆分開來,而不是把整個迭代部分放入一個組件中,使得這種模型處理迭代問題時更加靈活。

2 基于Storm模型設計的迭代方案

2.1 迭代Topology模型設計

基于Storm的基礎編程模型,文中改進其Topology,增加了迭代模塊,能夠解決迭代類的問題,如圖2所示。迭代Topology與之前Topology相比多了一個新拓撲:Iterator。Iterator負責處理具有迭代結構部分程序。里面同時包含了IBolt組件、迭代檢查器Checker和外部數(shù)據(jù)源接收器Receiver。IBolt組件負責任務處理,迭代檢查器負責判斷迭代是否結束,外部數(shù)據(jù)接收器負責接收迭代器外部發(fā)來的數(shù)據(jù)。迭代Topology好比在流水線上的工人中增加一個迭代管理員,負責管理處理迭代任務的工人,并告訴他們什么時候進行迭代操作,什么時候結束迭代。

圖2中描述的這種迭代模型看起來只能模仿Do-While循環(huán)模式,但While-Do或者Do-While在一定條件下可以互相轉化,所以這種模型其實可以解決任何迭代問題。

迭代 Topology新增了 Receiver、IBolt、Checker 3個組件,通過新增的 3個API(setReceiver()、setIBolt()、setChecker())來實現(xiàn)。

(1)新增組件

Receiver組件:Receiver組件用來接收外部組件和內部迭代組件Checker組件發(fā)送來的消息。將接收到的消息進行排隊處理后發(fā)往迭代開始的IBolt組件。Receiver組件一方面解決了Checker組件和Spout組件同時向IBolt組件發(fā)送消息的功能,另一方面也可以控制數(shù)據(jù)傳入IBolt組件的速率。

IBolt組件:IBolt組件與Bolt組件一樣,負責實際任務處理,但它處理的是需要迭代運算的任務,區(qū)別于非迭代功能的任務。

Checker組件:Checker組件是實現(xiàn)迭代過程的關鍵,迭代控制主要由它完成。主要功能是檢查迭代處理是否結束。與其它組件不同的是,有兩個發(fā)射口,一個是發(fā)往Receive組件,一個是發(fā)往外部Bolt組件。如果進入下次迭代,消息發(fā)往Receiver;如果迭代結束,消息發(fā)往外部Bolt組件。

圖2 迭代Topology

(2)新增組件API

setReceiver():構造1個接收器,2個參數(shù),參數(shù)1設定接收器名,參數(shù)2設定并行數(shù)目;

setIBolt():構造1個迭代處理組件,3個參數(shù),參數(shù)1設定迭代處理組件名,參數(shù)2設定要處理的任務,參數(shù)3設定并行數(shù)目;

setChecker():構造1個迭代檢查器,4個參數(shù),參數(shù)1設定迭代檢查器名,參數(shù)2設定消息接收器名,參數(shù)3設定迭代檢查器任務,參數(shù)4設定并行數(shù)目;

2.2 迭代Topology模型的實現(xiàn)

圖3為Storm中實現(xiàn)Topology模型的一種架構[4]。該架構由3個進程組成:Nimbus進程為主進程,負責接收客戶端提交的代碼,并將代碼序列化,為客戶機分發(fā)任務;Zookeeper進程負責Nimbus進程和Supervisor進程之間的消息協(xié)同工作;Supervisor進程負責接收任務并執(zhí)行任務,將任務結果返回給用戶。

實現(xiàn)這種迭代Topology,仍然延用Storm系統(tǒng)基礎架構,需要改變的是系統(tǒng)的調度策略、組件類型。一個完整迭代任務執(zhí)行過程如下:

(1)Nimbus進程接收客戶端提交過來的具有迭代結構的Topology,然后將每個組件序列化,并分發(fā)任務到Supervisor;

(2)Supervisor接收分發(fā)的任務并開始執(zhí)行;

(3)Spout組件發(fā)射一條消息到迭代消息接收器Receiver,消息接收器將消息進行排隊,按照先來先服務策略,把消息發(fā)往第一個要開始迭代操作的組件IBolt;

(4)IBolt組件處理完自己的任務,根據(jù)用戶設定,決定任務結果是發(fā)往Checker組件還是發(fā)往下一個IBolt組件;

(5)Checker組件根據(jù)接收到的消息和用戶定義的檢查點決定是否繼續(xù)迭代操作,如果迭代完成則將消息發(fā)往外部Bolt組件,否則,將消息發(fā)往Receiver組件開始下一次循環(huán)操作。

圖3 Storm框架

3 迭代Topology模型模擬實現(xiàn)K-Means算法

K-Means算法是數(shù)據(jù)挖掘中應用廣泛的數(shù)據(jù)聚類算法,算法程序結構包含了迭代處理部分,這種具有迭代結構的算法很多,為了驗證迭代Topology模型的可行性,文中選用比較典型并且為大家熟知的K-Means算法來實現(xiàn)。

K-Means算法的核心思想[5]是找出K個聚類中心c1,c2,…,ck,使每個數(shù)據(jù)點xi和與其最近的聚類中心cv的平方距離和被最小化(該平方距離和被稱為偏差D)。對n個樣本進行聚類的過程如下:

(1)初始化:隨機指定k個聚類中心(c1,c2,…,ck);

(2)重復下面過程直到D收斂:

①分配xi:對每個樣本xi,找到離它最近的聚類中心cv,并將其分配到cv所標明類;

②修正cv:對每一個cv移動到其標明的類中心;

藥士道：“此人之前乃是假死，蓋因體力透支嚴重，精神高度緊繃，再加之外傷失血過多，導致身體機能衰竭。如今能重新恢復氣息，實數(shù)罕見。容我為她配些養(yǎng)神滋補的草藥，至于她能不能徹底醒過來，還要看她自身的造化了?！?/p>

因此,在迭代 Topology模型上設計K-Means算法的Topology可以由圖4表示。

該Topology中,Spout組件為spout,Receiver組件為receiver,Checker組件為checker,IBolt處理組件包括caldistance和move,Bolt組件為writeResult,K-Means算法在迭代Topology上的實現(xiàn)過程的具體描述如圖5所示。

圖4 K-Means算法的迭代Topology

該實現(xiàn)采用5個點作為點群,選取2個種子點,圖5描述了這種歸類的大致過程,具體實現(xiàn)過程如下:

(1)spout組件隨機產生5個點作為點群,隨機產生2個點作為基點,然后將點群和基點作為數(shù)據(jù)輸入發(fā)往receiver組件;

(2)receiver組件接收到輸入后把輸入數(shù)據(jù)發(fā)往迭代開始組件caldistance,該組件用來計算基點與各個點群之間的距離;

(3)caldistance將計算完后的距離以及點群和基點位置發(fā)往move組件,move組件根據(jù)距離計算出點群中心,并將基點移動到點群中心,將移動后的基點與點群作為數(shù)據(jù)輸入發(fā)往checker組件;

(4)checker組件將第一次接收到的數(shù)據(jù)與初始化的基點相比較,比較完后將接收到的基點代替初始化的基點作為下一次比較的對象,初始化的基點位置一般為(0,0),因此第一次比較必然要進行迭代處理,所以checker組件將接收到的點群和基點發(fā)往receiver,然后開始第二次重復處理過程;

下面是構建K-Means迭代Topology的主要模擬實現(xiàn)代碼段:

TopologyBuilder builder=new TopologyBuilder();

builder.setSpout(“spout” ,new RandomProducePoints(5,2),2);

builder.setReceiver(“receiver” ,4);

builder.setIBolt(“caldistance” ,new calDistance(2),2).shufflegrouping(“ receiver”);

builder.setIBolt(“move”,new MovetoCenter(),2).shufflegrouping(“caldistance”);

builder.setChecker(“checker” ,”receiver” ,new checkChanges(),2).shufflegrouping(“move”);

builder.setBolt(“writeResult”,new writeResult(),1).shufflegrouping(“checker”);

以上代碼都是模擬仿真Storm系統(tǒng)中實現(xiàn)Topology的代碼來表述的。TopologyBuilder類是構建一個Topology圖需要的類,類中包含了設置各種組件的方法,一個應用的Topology就是通過這個類來實現(xiàn)的,組件中的參數(shù)主要有4類:組件名;組件任務;接收消息的方式;并行數(shù)目。接受消息有6種分組方式:隨機分組、字段分組、全部分組、全局分組、無分組、直接分組,示例中都為隨機分組。隨機分組有一個參數(shù),這個參數(shù)指定接受消息的組件名。隨機分組的意思就是隨機接受組件發(fā)送來的消息。例如,組件move有4個線程在同時工作,組件caldistance有5個線程同時工作,move隨機接收caldistance發(fā)送來的消息,就表示move的任意一個線程接受caldistance任意線程計算完后的結果。并行數(shù)目,指共同執(zhí)行該組件任務的線程數(shù)目。并行數(shù)目都需要用戶根據(jù)任務情況來設置。示例中receiver采用4個線程,writeResult采用1個線程,其它的都是2個線程。receiver線程數(shù)目為4,因為它既要接收spout的消息還要接收checker的消息。

圖5 迭代 Topology實現(xiàn)K-Means算法過程

4 結束語

迭代 Topology在Storm Topology原型基礎上增加了 Receiver、IBolt、Checker組件,Receiver和Checker組件與IBolt組件連接組成了一個具有迭代功能的Topology圖,使用這種迭代Topology圖成功解決了具有迭代功能的K-Means算法,這種方案因為是在以前基礎上添加組件完成,所以就很好保留了原Topology的特點。這種迭代Topology在實現(xiàn)方式上,保留了Storm基礎架構,只是調整了主進程的調度策略和組件類型,實現(xiàn)上降低了后續(xù)開發(fā)難度。

致謝:感謝成都市科技局創(chuàng)新發(fā)展戰(zhàn)略研究項目(11RKYB016ZF)對本文的資助

[1] 孟小峰,慈祥.大數(shù)據(jù)管理:概念,技術與挑戰(zhàn)[J].計算機研究與發(fā)展,2013,50(1):146-169.

[2] Ekanayake J,Li H,Zhang B,et al.Twister:a runtime for iterative mapreduce[C].Proceedingsof the 19th ACM International Symposium on High Performance Distributed Computing.ACM,2010:810-818.

[3] Bu Y,Howe B,Balazinska M,et al.HaLoop:Efficient iterative data processing on large clusters[J].Proceedings of the VLDB Endowment,2010,3(1-2):285-296.

[4] Storm-wiki.[EB/OL].http://github.com/nathanmarz/storm/wiki/,2013-06-10.

[5] 孫吉貴,劉杰,趙連宇.聚類算法研究[J].軟件學報,2008,19(1):48-61.

[6] Dean J,Ghemawat S.MapReduce:simplified data processing on large clusters[J].Communications of the ACM,2008,51(1):107-113.

[7] Neumeyer L,Robbins B,Nair A,et al.S4:Distributed stream computing platform[C].Data Mining Workshops(ICDMW),2010 IEEE International Conference on.IEEE,2010:170-177.

[8] Cherniack M,Balakrishnan H,Balazinska M,et al.Scalable Distributed Stream Processing[C].CIDR.2003,3:257-268.

[9] Nathanmarz-blog[EB/OL].http://nathanmarz.com/.2013-07-10.

[10] Storm-berkeley[EB/OL].storm-berkeley.pdf.2013-09-01.

[11] 張建萍,劉希玉.基于聚類分析的K-means算法研究及應用[J].計算機應用研究,2007,24(5):166-168.

[12] 鄧華鋒,劉云生,肖迎元.分布式數(shù)據(jù)流處理系統(tǒng)的動態(tài)負載平衡技術[J].計算機科學,2007,34(7):120-123.

[13] 亓開元,趙卓峰,房俊,等.針對高速數(shù)據(jù)流的大規(guī)模數(shù)據(jù)實時處理方法[J].計算機學報,2012,35(3):477-490.

[14] Getting Started with Storm[EB/OL].Getting Started with Storm.pdf.2013-08-18.

[15] S4 vs Storm[EB/OL].s4vStorm.pdf.2013-08-08.