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

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

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

　　關鍵詞：實時計算；Storm；并行編程

0引言

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

1Storm簡介

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

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

2實時數據收集系統的設計

　　由于傳統的數據收集與處理通常是將文件離線存儲在硬盤上，再進行處理，實時性不高。為了解決這個問題，基于Storm開發一個實時數據收集與處理的系統，該系統直接將數據轉為流的形式進行處理，而不再存儲到文件，從而實現高響應低延遲。

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

　　2.1Flume

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

　　2.2Kafka

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

　　2.3Storm工作機制

　　Storm由主節點和從節點構成。主節點運行Nimbus進程，負責代碼的分發等分配任務。從節點Supervisor負責接收主節點分給它的任務，一般情況下，一個從節點上運行一個或多個進程Worker，每個進程中又產生一個或多個線程Excutor，線程用來執行Task任務，即Spout和Bolt業務邏輯。

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

3系統實現

　　3.1數據源

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

　　通過內存中產生數據模擬實時數據，以log4j形式進行接收，Flume用avro網絡流的形式讀取log4j接收到的數據。采用XML配置集成log4j與Flume，完成數據源source對實時數據的讀取。圖2是Storm工作流程圖。

　　3.2Flume集成Kafka

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

　　3.3Storm處理數據流

　　Spout是一個Topology的數據源，在服務器上應該時刻在產生數據，Spout讀取實時的數據。本文將數據以log4j的形式在內存中不斷產生，根據設計的系統，Spout將從Kafka中不斷讀取數據。這里不需要手動實現一個Spout，stormkafka的插件中已經提供好一個類KafkaSpout，使用它Storm會從Kafka中不斷讀數據。

　　數據流流入bolt進行邏輯處理，Spout產生的話題作為關鍵字，對源碼進行分析可知，Bolt會根據“bytes”字段將接收過來的數據進行區分。根據設計的系統流程，每當產生一個話題，就向關系型數據庫插入一條數據，用來持久化。

　　3.4系統性能測試

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

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

　　3.5Storm并行度測試

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

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

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

4結論

　　本文通過對Storm的體系結構進行分析，結合Flume、Kafka框架，實現了一個實時數據收集與計算的系統。該系統具有良好的可擴展性。下一步的主要工作是接入實際生產中的數據源，如實時股票浮動數據、實時天氣數據等，并且對Storm的實時計算結果進行動態的可視化。

參考文獻

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

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

　　［3］ 孫大為,張廣艷,鄭緯民.大數據流式計算：關鍵技術及系統實例［J］.軟件學報,2014,25(4):839862.

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

　　［5］ 鄧立龍,徐海水.Storm實現的應用模型研究［J］.廣東工業大學學報,2014,31(3):114118.

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

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