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

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

　　摘要：大數(shù)據(jù)計(jì)算主要包括批量計(jì)算和實(shí)時(shí)計(jì)算，針對批量計(jì)算處理的都是歷史數(shù)據(jù)而不能實(shí)現(xiàn)低延遲高響應(yīng)的缺點(diǎn)，基于Storm并借助開源框架Flume、Kafka，設(shè)計(jì)了一個(gè)實(shí)時(shí)數(shù)據(jù)收集與處理的系統(tǒng)，將數(shù)據(jù)轉(zhuǎn)為流的形式，對收集來的數(shù)據(jù)直接在內(nèi)存以流的形式進(jìn)行計(jì)算，輸出有價(jià)值的信息。最后對系統(tǒng)進(jìn)行性能測試以及計(jì)算能力的測試。實(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)注的熱門話題。有國家指出，大數(shù)據(jù)就是“未來的新石油”，擁有數(shù)據(jù)的規(guī)模和運(yùn)用數(shù)據(jù)的能力將會成為一個(gè)國家綜合國力的重要組成部分［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í)數(shù)據(jù)方面，Hadoop卻顯得有些力不從心。由此，Twitter公司研發(fā)出開源的實(shí)時(shí)數(shù)據(jù)計(jì)算技術(shù)Storm，它的出現(xiàn)使得大數(shù)據(jù)的實(shí)時(shí)處理成為可能。

1Storm簡介

　　Storm是一個(gè)分布式的實(shí)時(shí)計(jì)算系統(tǒng)，它可以方便地在一個(gè)計(jì)算機(jī)集群中編寫復(fù)雜的實(shí)時(shí)計(jì)算。Storm之于實(shí)時(shí)處理，就好比Mapreduce批處理［4］。其主要應(yīng)用場景有流數(shù)據(jù)處理、實(shí)時(shí)搜索等［5］。Storm保證每個(gè)消息都會得到處理，而且它很快——在一個(gè)小集群中，每秒可以處理數(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用于一個(gè)Topology拓?fù)渖a(chǎn)消息，一般是從外部數(shù)據(jù)源接入，將數(shù)據(jù)流以tuple的形式傳遞給Bolt，Bolt去處理傳過來的tuple。

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

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

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

　　2.1Flume

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

　　2.2Kafka

　　Kafka是一個(gè)分布式發(fā)布訂閱消息系統(tǒng)。核心組件包括producer、broker和consumer。一個(gè)單一的Kafka的broker可以處理數(shù)百兆字節(jié)讀取和寫入每秒從成千上萬的客戶。Kafka內(nèi)部的消息隊(duì)列可以實(shí)現(xiàn)高吞吐量、高負(fù)載，當(dāng)數(shù)據(jù)源頭數(shù)據(jù)量過大時(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，線程用來執(zhí)行Task任務(wù)，即Spout和Bolt業(yè)務(wù)邏輯。

　　當(dāng)一個(gè)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中。整個(gè)過程主節(jié)點(diǎn)與從節(jié)點(diǎn)之間是不進(jìn)行通信的，所有的通信通過zk來協(xié)調(diào)［6］。

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

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

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

　　通過內(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對實(shí)時(shí)數(shù)據(jù)的讀取。圖2是Storm工作流程圖。

　　3.2Flume集成Kafka

　　雖然Flume已經(jīng)自帶實(shí)現(xiàn)了avro sink、hdfs sink等多種sink，但是沒有對接Kafka的sink，因此，首先自定義流向到Kafka的sink。通過繼承一個(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會從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)生一個(gè)話題，就向關(guān)系型數(shù)據(jù)庫插入一條數(shù)據(jù)，用來持久化。

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

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

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

　　3.5Storm并行度測試

　　為了測試Storm橫向擴(kuò)展給整個(gè)系統(tǒng)帶來的性能影響，分別在單個(gè)節(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ù)量，此時(shí)隨并行度的增大計(jì)算能力會下降。無論是單節(jié)點(diǎn)還是集群，Storm的計(jì)算能力都會隨線程數(shù)量的增加而增強(qiáng)。

4結(jié)論

　　本文通過對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ù)等，并且對Storm的實(shí)時(shí)計(jì)算結(jié)果進(jìn)行動(dòng)態(tài)的可視化。

參考文獻(xiàn)

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

　?。?］ 覃雄派,王會舉,杜小勇，等.大數(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.

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

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