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

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

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

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

0引言

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

1Storm簡(jiǎn)介

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

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

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

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

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

　　2.1Flume

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

　　2.2Kafka

　　Kafka是一個(gè)分布式發(fā)布訂閱消息系統(tǒng),。核心組件包括producer、broker和consumer,。一個(gè)單一的Kafka的broker可以處理數(shù)百兆字節(jié)讀取和寫(xiě)入每秒從成千上萬(wàn)的客戶,。Kafka內(nèi)部的消息隊(duì)列可以實(shí)現(xiàn)高吞吐量、高負(fù)載,，當(dāng)數(shù)據(jù)源頭數(shù)據(jù)量過(guò)大時(shí),，可以在Kafka內(nèi)部得到有效的暫存，否則,，Storm集群處理數(shù)據(jù)的速度可能趕不上數(shù)據(jù)源數(shù)據(jù)流動(dòng)的速度,，容易造成集群的性能減慢甚至宕機(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ù)，一般情況下,，一個(gè)從節(jié)點(diǎn)上運(yùn)行一個(gè)或多個(gè)進(jìn)程Worker,，每個(gè)進(jìn)程中又產(chǎn)生一個(gè)或多個(gè)線程Excutor，線程用來(lái)執(zhí)行Task任務(wù),，即Spout和Bolt業(yè)務(wù)邏輯,。

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

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

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

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

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

　　3.2Flume集成Kafka

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

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

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

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

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

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

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

　　3.5Storm并行度測(cè)試

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

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

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

4結(jié)論

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

參考文獻(xiàn)

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

　?。?］ 覃雄派,王會(huì)舉,杜小勇,，等.大數(shù)據(jù)分析——RDBMS與MapReduce的競(jìng)爭(zhēng)與共生［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.

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

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

　?。?］ 王銘坤,袁少光,朱永利,，等.基于Storm的海量數(shù)據(jù)實(shí)時(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.