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

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

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

　　關(guān)鍵詞：實時計算；Storm；并行編程

0引言

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

1Storm簡介

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

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

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

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

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

　　2.1Flume

　　Flume是一個分布式的、能有效地收集、移動大量日志數(shù)據(jù)的框架。它有一個簡單的和靈活的基于流的數(shù)據(jù)流架構(gòu)。這是強大的容錯的可靠性機制和多故障轉(zhuǎn)移和恢復(fù)機制。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)部的消息隊列可以實現(xiàn)高吞吐量、高負載，當(dāng)數(shù)據(jù)源頭數(shù)據(jù)量過大時，可以在Kafka內(nèi)部得到有效的暫存，否則，Storm集群處理數(shù)據(jù)的速度可能趕不上數(shù)據(jù)源數(shù)據(jù)流動的速度，容易造成集群的性能減慢甚至宕機。

　　2.3Storm工作機制

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

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

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

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

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

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

　　3.2Flume集成Kafka

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

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

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

　　數(shù)據(jù)流流入bolt進行邏輯處理，Spout產(chǎn)生的話題作為關(guān)鍵字，對源碼進行分析可知，Bolt會根據(jù)“bytes”字段將接收過來的數(shù)據(jù)進行區(qū)分。根據(jù)設(shè)計的系統(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的計算能力，當(dāng)海量數(shù)據(jù)產(chǎn)生時，甚至可能導(dǎo)致Storm的宕機。為了測試整個系統(tǒng)中Kafka的消息隊列是否能暫存大量的數(shù)據(jù)，采用多線程［7］技術(shù)，同時運行若干線程產(chǎn)生數(shù)據(jù)，每個線程產(chǎn)生相同的數(shù)據(jù)量，通過對Storm計算效率的對比來驗證。表1為Storm計算性能與線程數(shù)的關(guān)系。圖3為Storm處理數(shù)據(jù)量與時間的關(guān)系。

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

　　3.5Storm并行度測試

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

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

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

4結(jié)論

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

參考文獻

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

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

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

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

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

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

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