趙菲1,，林穗1,，高西剛2

　　(1 廣東工業(yè)大學(xué) 計(jì)算機(jī)學(xué)院，廣東 廣州 510006,；2 廣東訊飛啟明科技發(fā)展有限公司,，廣東 廣州 510530)

　　摘要：大數(shù)據(jù)計(jì)算主要包括批量計(jì)算和實(shí)時計(jì)算，針對批量計(jì)算處理的都是歷史數(shù)據(jù)而不能實(shí)現(xiàn)低延遲高響應(yīng)的缺點(diǎn),，基于Storm并借助開源框架Flume,、Kafka，設(shè)計(jì)了一個實(shí)時數(shù)據(jù)收集與處理的系統(tǒng),，將數(shù)據(jù)轉(zhuǎn)為流的形式,，對收集來的數(shù)據(jù)直接在內(nèi)存以流的形式進(jìn)行計(jì)算,，輸出有價值的信息。最后對系統(tǒng)進(jìn)行性能測試以及計(jì)算能力的測試,。實(shí)驗(yàn)結(jié)果表明,，該系統(tǒng)可擴(kuò)展性良好，且并行計(jì)算能力穩(wěn)定,，適合大量實(shí)時數(shù)據(jù)處理,。

　　關(guān)鍵詞：實(shí)時計(jì)算；Storm,；并行編程

0引言

　　近些年,，大數(shù)據(jù)已迅速成為科技界、學(xué)術(shù)界關(guān)注的熱門話題,。有國家指出,，大數(shù)據(jù)就是“未來的新石油”，擁有數(shù)據(jù)的規(guī)模和運(yùn)用數(shù)據(jù)的能力將會成為一個國家綜合國力的重要組成部分［1］,?；ヂ?lián)網(wǎng)、社交媒體等眾多行業(yè)領(lǐng)域每天都產(chǎn)生GB甚至TB的數(shù)據(jù),。隨著數(shù)據(jù)量的指數(shù)級增長,，也催生了一批大數(shù)據(jù)處理技術(shù)。目前,，最流行的莫過于Hadoop分布式存儲計(jì)算平臺［2］,。Hadoop適合對數(shù)據(jù)進(jìn)行批量處理，可以應(yīng)用到日志分析,、建立索引等方面［3］,。這些都是對歷史數(shù)據(jù)進(jìn)行集中處理，但是在對待實(shí)時數(shù)據(jù)方面,，Hadoop卻顯得有些力不從心,。由此，Twitter公司研發(fā)出開源的實(shí)時數(shù)據(jù)計(jì)算技術(shù)Storm,，它的出現(xiàn)使得大數(shù)據(jù)的實(shí)時處理成為可能,。

1Storm簡介

　　Storm是一個分布式的實(shí)時計(jì)算系統(tǒng)，它可以方便地在一個計(jì)算機(jī)集群中編寫復(fù)雜的實(shí)時計(jì)算,。Storm之于實(shí)時處理,，就好比Mapreduce批處理［4］。其主要應(yīng)用場景有流數(shù)據(jù)處理,、實(shí)時搜索等［5］,。Storm保證每個消息都會得到處理，而且它很快——在一個小集群中，每秒可以處理數(shù)以百萬計(jì)的消息,。

　　Storm實(shí)現(xiàn)了一種流式處理模型,，流是一種有順序并且連續(xù)到達(dá)的數(shù)據(jù),。在Storm計(jì)算模型中,，主要有兩類計(jì)算過程，分別是Spout計(jì)算過程和Bolt計(jì)算過程,。Spout,、Bolt分別由用戶自己實(shí)現(xiàn)SpoutInterface和BoltInterface兩類接口。Spout用于一個Topology拓?fù)渖a(chǎn)消息,，一般是從外部數(shù)據(jù)源接入,，將數(shù)據(jù)流以tuple的形式傳遞給Bolt，Bolt去處理傳過來的tuple,。

2實(shí)時數(shù)據(jù)收集系統(tǒng)的設(shè)計(jì)

　　由于傳統(tǒng)的數(shù)據(jù)收集與處理通常是將文件離線存儲在硬盤上,，再進(jìn)行處理，實(shí)時性不高,。為了解決這個問題,，基于Storm開發(fā)一個實(shí)時數(shù)據(jù)收集與處理的系統(tǒng)，該系統(tǒng)直接將數(shù)據(jù)轉(zhuǎn)為流的形式進(jìn)行處理,，而不再存儲到文件,，從而實(shí)現(xiàn)高響應(yīng)低延遲。

　　為了保證整個系統(tǒng)的健壯性及可維護(hù)性,，采取Flume和Kafka框架對實(shí)時數(shù)據(jù)進(jìn)行收集與緩存,，Storm則對實(shí)時數(shù)據(jù)流進(jìn)行處理。圖1為整個系統(tǒng)的架構(gòu)圖,。

　　2.1Flume

　　Flume是一個分布式的,、能有效地收集、移動大量日志數(shù)據(jù)的框架,。它有一個簡單的和靈活的基于流的數(shù)據(jù)流架構(gòu),。這是強(qiáng)大的容錯的可靠性機(jī)制和多故障轉(zhuǎn)移和恢復(fù)機(jī)制。Flume的核心組件中包含source,、channel,，sink。source允許快速接入各種數(shù)據(jù)源,，數(shù)據(jù)經(jīng)過channel管道的暫存,，最終sink使得數(shù)據(jù)流向多種目的地。

　　2.2Kafka

　　Kafka是一個分布式發(fā)布訂閱消息系統(tǒng),。核心組件包括producer,、broker和consumer。一個單一的Kafka的broker可以處理數(shù)百兆字節(jié)讀取和寫入每秒從成千上萬的客戶。Kafka內(nèi)部的消息隊(duì)列可以實(shí)現(xiàn)高吞吐量,、高負(fù)載,，當(dāng)數(shù)據(jù)源頭數(shù)據(jù)量過大時，可以在Kafka內(nèi)部得到有效的暫存,，否則,，Storm集群處理數(shù)據(jù)的速度可能趕不上數(shù)據(jù)源數(shù)據(jù)流動的速度，容易造成集群的性能減慢甚至宕機(jī),。

　　2.3Storm工作機(jī)制

　　Storm由主節(jié)點(diǎn)和從節(jié)點(diǎn)構(gòu)成,。主節(jié)點(diǎn)運(yùn)行Nimbus進(jìn)程，負(fù)責(zé)代碼的分發(fā)等分配任務(wù),。從節(jié)點(diǎn)Supervisor負(fù)責(zé)接收主節(jié)點(diǎn)分給它的任務(wù),，一般情況下，一個從節(jié)點(diǎn)上運(yùn)行一個或多個進(jìn)程Worker,，每個進(jìn)程中又產(chǎn)生一個或多個線程Excutor,，線程用來執(zhí)行Task任務(wù)，即Spout和Bolt業(yè)務(wù)邏輯,。

　　當(dāng)一個Topology被定義好后,，通過Storm的jar命令，將它打包上傳至主節(jié)點(diǎn),，主節(jié)點(diǎn)去zk檢查集群的狀態(tài)是否處于active,，并且檢查集群中是否存在相同名字的Topology實(shí)例在運(yùn)行，然后根據(jù)代碼中的參數(shù)來確定進(jìn)程,、線程,、Spout和Bolt的數(shù)量，并設(shè)定Task任務(wù)的數(shù)量以及Task編號,，將分配好的Task信息寫入zk中,。整個過程主節(jié)點(diǎn)與從節(jié)點(diǎn)之間是不進(jìn)行通信的，所有的通信通過zk來協(xié)調(diào)［6］,。

3系統(tǒng)實(shí)現(xiàn)

　　3.1數(shù)據(jù)源

　　新浪微博作為時下流行的社交平臺,，每時每刻都會產(chǎn)生不同的熱門話題以反映民眾的關(guān)注點(diǎn)。為了統(tǒng)計(jì)實(shí)時的熱門話題內(nèi)容,，采用在內(nèi)存中不定時隨機(jī)產(chǎn)生不同的詞條的方法來模擬微博上的話題,，并且通過Storm框架對數(shù)據(jù)進(jìn)行分析統(tǒng)計(jì)出實(shí)時的熱門話題。

　　通過內(nèi)存中產(chǎn)生數(shù)據(jù)模擬實(shí)時數(shù)據(jù),，以log4j形式進(jìn)行接收,，F(xiàn)lume用avro網(wǎng)絡(luò)流的形式讀取log4j接收到的數(shù)據(jù)。采用XML配置集成log4j與Flume,，完成數(shù)據(jù)源source對實(shí)時數(shù)據(jù)的讀取,。圖2是Storm工作流程圖,。

　　3.2Flume集成Kafka

　　雖然Flume已經(jīng)自帶實(shí)現(xiàn)了avro sink、hdfs sink等多種sink,，但是沒有對接Kafka的sink,，因此，首先自定義流向到Kafka的sink,。通過繼承一個AbstrackSink并且實(shí)現(xiàn)Configurable接口,。其中有兩個核心方法，分別是configure()和process(),，實(shí)現(xiàn)自定義Flume到Kafka的sink,。

　　3.3Storm處理數(shù)據(jù)流

　　Spout是一個Topology的數(shù)據(jù)源,，在服務(wù)器上應(yīng)該時刻在產(chǎn)生數(shù)據(jù),，Spout讀取實(shí)時的數(shù)據(jù)。本文將數(shù)據(jù)以log4j的形式在內(nèi)存中不斷產(chǎn)生,，根據(jù)設(shè)計(jì)的系統(tǒng),，Spout將從Kafka中不斷讀取數(shù)據(jù)。這里不需要手動實(shí)現(xiàn)一個Spout,，stormkafka的插件中已經(jīng)提供好一個類KafkaSpout,，使用它Storm會從Kafka中不斷讀數(shù)據(jù)。

　　數(shù)據(jù)流流入bolt進(jìn)行邏輯處理,，Spout產(chǎn)生的話題作為關(guān)鍵字,，對源碼進(jìn)行分析可知，Bolt會根據(jù)“bytes”字段將接收過來的數(shù)據(jù)進(jìn)行區(qū)分,。根據(jù)設(shè)計(jì)的系統(tǒng)流程,，每當(dāng)產(chǎn)生一個話題，就向關(guān)系型數(shù)據(jù)庫插入一條數(shù)據(jù),，用來持久化,。

　　3.4系統(tǒng)性能測試

　　當(dāng)數(shù)據(jù)源不停產(chǎn)生數(shù)據(jù)時，可能會削弱Storm的計(jì)算能力,，當(dāng)海量數(shù)據(jù)產(chǎn)生時,，甚至可能導(dǎo)致Storm的宕機(jī)。為了測試整個系統(tǒng)中Kafka的消息隊(duì)列是否能暫存大量的數(shù)據(jù),，采用多線程［7］技術(shù),，同時運(yùn)行若干線程產(chǎn)生數(shù)據(jù)，每個線程產(chǎn)生相同的數(shù)據(jù)量,，通過對Storm計(jì)算效率的對比來驗(yàn)證,。表1為Storm計(jì)算性能與線程數(shù)的關(guān)系。圖3為Storm處理數(shù)據(jù)量與時間的關(guān)系,。

　　通過上述圖表可以清晰地看出,，Storm的計(jì)算時間是隨著數(shù)據(jù)量的增大線性遞增的，而不是指數(shù)型遞增，說明Storm的計(jì)算能力沒有隨著數(shù)據(jù)量的增大而削弱,，在Kafka的作用下,，數(shù)據(jù)得到了有效的緩存，使得Storm正常運(yùn)作,，從而保證整個系統(tǒng)的穩(wěn)定性,。

　　3.5Storm并行度測試

　　為了測試Storm橫向擴(kuò)展給整個系統(tǒng)帶來的性能影響，分別在單個節(jié)點(diǎn)和集群的情況下進(jìn)行測試,。對paramhint參數(shù)設(shè)置,，改變進(jìn)程、線程數(shù)量,，分析Storm計(jì)算能力,。

　　通過圖4、圖5可以得出,，在單節(jié)點(diǎn)下,，隨著線程數(shù)量的增多Storm計(jì)算能力反而減弱；而在集群下,，隨著線程的增加計(jì)算能力增加,。所以建議在非集群下，不要增加進(jìn)　

　程的數(shù)量,，此時隨并行度的增大計(jì)算能力會下降,。無論是單節(jié)點(diǎn)還是集群，Storm的計(jì)算能力都會隨線程數(shù)量的增加而增強(qiáng),。

4結(jié)論

　　本文通過對Storm的體系結(jié)構(gòu)進(jìn)行分析,，結(jié)合Flume、Kafka框架,，實(shí)現(xiàn)了一個實(shí)時數(shù)據(jù)收集與計(jì)算的系統(tǒng),。該系統(tǒng)具有良好的可擴(kuò)展性。下一步的主要工作是接入實(shí)際生產(chǎn)中的數(shù)據(jù)源,，如實(shí)時股票浮動數(shù)據(jù),、實(shí)時天氣數(shù)據(jù)等，并且對Storm的實(shí)時計(jì)算結(jié)果進(jìn)行動態(tài)的可視化,。

參考文獻(xiàn)

　?。?］ 程學(xué)旗,靳小龍,王元卓,等.大數(shù)據(jù)系統(tǒng)和分析技術(shù)綜述［J］. 軟件學(xué)報(bào),2014,25(9):18891908.

　　［2］ 覃雄派,王會舉,杜小勇,，等.大數(shù)據(jù)分析——RDBMS與MapReduce的競爭與共生［J］.軟件學(xué)報(bào),2012,23(1):3245.

　?。?］ 孫大為,張廣艷,鄭緯民.大數(shù)據(jù)流式計(jì)算：關(guān)鍵技術(shù)及系統(tǒng)實(shí)例［J］.軟件學(xué)報(bào),2014,25(4):839862.

　　［4］ 鄭曉薇,馬琳.基于Hadoop集群的多表并行關(guān)聯(lián)算法及應(yīng)用［J］.微型機(jī)與應(yīng)用,2013,32(4):9193.

　?。?］ 鄧立龍,徐海水.Storm實(shí)現(xiàn)的應(yīng)用模型研究［J］.廣東工業(yè)大學(xué)學(xué)報(bào),2014,31(3):114118.

　?。?］ 王銘坤,袁少光,朱永利,，等.基于Storm的海量數(shù)據(jù)實(shí)時聚類［J］.計(jì)算機(jī)應(yīng)用,2014,34(11):30783081.

　　［7］ 王維,李濤,韓俊剛.一種多線程輕核機(jī)器中進(jìn)程管理的硬件實(shí)現(xiàn)［J］.電子技術(shù)應(yīng)用,2013,39(3):4043.