一種基于Storm編程模型的迭代Topology方案

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

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

0 引言

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

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

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

1 問題描述

1.1 Storm編程模型思想

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

圖1 Storm Topology

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

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

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

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

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

2.1 迭代Topology模型設計

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

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

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

(1)新增組件

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

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

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

圖2 迭代Topology

(2)新增組件API

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

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

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

2.2 迭代Topology模型的實現

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

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

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

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

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

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

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

圖3 Storm框架

3 迭代Topology模型模擬實現K-Means算法

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

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

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

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

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

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

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

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

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

圖4 K-Means算法的迭代Topology

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

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

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

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

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

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

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系統中實現Topology的代碼來表述的。TopologyBuilder類是構建一個Topology圖需要的類,類中包含了設置各種組件的方法,一個應用的Topology就是通過這個類來實現的,組件中的參數主要有4類:組件名;組件任務;接收消息的方式;并行數目。接受消息有6種分組方式:隨機分組、字段分組、全部分組、全局分組、無分組、直接分組,示例中都為隨機分組。隨機分組有一個參數,這個參數指定接受消息的組件名。隨機分組的意思就是隨機接受組件發送來的消息。例如,組件move有4個線程在同時工作,組件caldistance有5個線程同時工作,move隨機接收caldistance發送來的消息,就表示move的任意一個線程接受caldistance任意線程計算完后的結果。并行數目,指共同執行該組件任務的線程數目。并行數目都需要用戶根據任務情況來設置。示例中receiver采用4個線程,writeResult采用1個線程,其它的都是2個線程。receiver線程數目為4,因為它既要接收spout的消息還要接收checker的消息。

圖5 迭代 Topology實現K-Means算法過程

4 結束語

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

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

[1] 孟小峰,慈祥.大數據管理:概念,技術與挑戰[J].計算機研究與發展,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] 鄧華鋒,劉云生,肖迎元.分布式數據流處理系統的動態負載平衡技術[J].計算機科學,2007,34(7):120-123.

[13] 亓開元,趙卓峰,房俊,等.針對高速數據流的大規模數據實時處理方法[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.