基于Storm的城市消防聯網遠程監控系統的實時數據處理應用

楊素素

(首都師范大學 信息工程學院,北京 100048)

基于Storm的城市消防聯網遠程監控系統的實時數據處理應用

楊素素

(首都師范大學 信息工程學院,北京 100048)

針對城市消防聯網遠程監控系統中實時信息數據逐漸增長而引出的大數據問題，傳統的消防系統無法實時、高效地處理消防實時數據的問題，提出了一種基于云計算和Storm實時數據處理系統的解決方案;對于開源的Storm框架進行需求和性能分析，實現對其技術架構上的改進，并結合消防系統的特點，提出一套高實時性、高可擴展性的消防聯網監控中心的數據實時處理的體系架構，同時也進行了云計算平臺的搭建，利用心跳檢測機制保證各個監控單位的實時性連接;研究表明，基于云計算和Storm平臺架構完全適用于消防聯網監控中心的實時消防數據的處理，具有高效性、高可靠性、性能顯著等特性。

Storm;消防預警;大數據實時處理;云計算

0 引言

近年來，各種火災事故頻發，有交通工具火災事故、生產生活中的火災事故、工業生產中的火災事故。80%以上的這些火災事故給人們帶來很慘重的損失。據分析調查發現，很多我們生活中常見的火災事故都是可以預先發現，并及時處理和疏散來減少損失的。這就對于火災危險性進行預警分析和評價是十分重要的。

為了解決火災及時報警、設備故障及時發現及時巡檢，提高城市的消防減災能力，各個城市都在努力建設消防聯網平臺。消防聯網覆蓋的范圍越來越廣，消防檢測設備的監控探頭和煙霧感應、溫度感應檢測等相關設備的檢測數據量也逐漸增大，尤其是消防聯網傳感設備中的采集的數據對于火災的預判和分析來說越來越重要，各個地區的上百萬臺設備傳輸過來的海量數據的處理也越來越迫切，大數據的實時處理問題也越來越突出。

現有的城市消防聯網遠程監控系統主要研究的是對消防數據的采集、以及采用更先進的傳輸設備和更可靠的傳輸協議對數據進行傳輸以保證數據的可靠傳輸。然而系統地解決消防預警困難，不僅僅需要數據的實時可靠傳輸，我們還需要對這些實時傳感數據進行實時地分析和預測，實時監控、遠程查看和遠程控制，以達到火災的及時預警、及時指揮、消防設備故障及時發現和整修。

現有的大數據處理方式分為流處理[1-3]和批處理兩種，數據批處理主要是針對靜態數據和變化比較慢的一些數據集的處理。對于高實時性要求的系統，特別是為了應對不斷增長的數據，如消防的檢測設備中實時檢測到異常數據、設備每時每刻的工作狀態的判斷等都要求我們更多地關注流數據的實時處理。然而，現如今，現有的顯著的流處理計算系統有Linkedin的Kalka[4]系統、Twitter的Storm[5]系統、Yahoo的S4[6](simple scalable streaming system)系統、Facebook的Data Freeway and Puma[7]系統、MillWheel[8]、Spark Streaming[9]和Photon[10]等，流處理技術已經作為傳統數據庫成品輸送管道中很重要的一部分[11-13]。本文主要在消防監控中心采用基于云計算平臺和Storm實時數據處理的系統設計，并對系統整體性能進行測試。

1 適用于城市消防聯網遠程監控系統的云計算平臺

目前消防行業里，隨著科技技術的迅猛發展，各種各樣具備先進性能的探測器和消防設備層出不窮，消防單位內所安裝設備的種類大大豐富，所產生的信息量也是爆發式的增長，一個普通中等城市的數據量已經達到了很大的規模，而且隨著城市建設的推進，數據量必然會達到大數據級別。城市消防遠程監控系統是將監控中心與各個消防單位之間建立網絡通訊，并結合運用地理信息系統、數字視頻監控等現代網絡信息技術，在監控中心內對所有在此聯網內的建筑物的火災報警情況進行實時監控、監測、對設備的運行情況進行集中監督和管理。同時國標要求：地級市以上城市(含地級,全國共286個)，應設立城市消防遠程監控系統；縣級城市宜設立城市消防遠程監控系統。

1.1 目前城市消防聯網遠程監控系統所受到的限制

目前我國的城市消防聯網遠程監控系統還處于廠家各自建設階段。通常消防單位中消防監控系統的信息數據存儲在各自的數據庫中，城市消防控制中心對各個監控單位的數據庫進行輪詢，然后對查詢到的數據進行處理，最后顯示在頁面上。這種方式在面對海量大數據的時候就顯得力不從心，而且存在諸多問題：

(1)效率低：當前大多數的城市消防聯網遠程監控系統，都是消防單位將各自的數據存儲在自己的數據庫中，城市消防監控中心通過不斷的查詢各個監控單位的數據庫來獲取數據，然后對數據進行處理，最后顯示在頁面上。隨著監控單位數據庫數據量的不斷增加，數據庫的檢索效率也會不斷降低，系統效率很難確保。

(2)實時性差：由于當前的城市消防聯網遠程監控系統的效率低，消防中心得到數據的實時性就無法確保，至少會有幾十秒左右的延遲，這幾十秒的時間對于火災預警來說至關重要，甚至關乎人的生命。對于消防監控中心來說數據的實時性是需要解決的最大的問題。

(3)維護難度大：當前的城市消防聯網遠程監控系統需要消防監控中心去不斷的查詢消防單位的數據庫，從而給各個節點上的服務器都帶來很大的壓力，容易導致宕機。同時，系統的容錯能力也較差，實際使用中系統的維護難度較大。

因此，高實時性、高效率、高性能的云計算集群架構的搭建是解決以上問題的最好方式，適用于城市消防聯網遠程監控系統的云計算平臺的搭建是刻不容緩的。

1.2 城市消防聯網遠程監控系統的云計算平臺的構建

本系統采用如圖1所示的系統框架，根據消防行業的實際情況，消防單位之間沒有必要建立數據連接，每個消防單位都直接和城市消防監控中心建立連接。采用心跳檢測機制來保障中心實時地監控消防單位的消防系統中主要服務器的運行狀況，消防監控中心實時采集數據并通過服務器集群進行高效分析和計算，最后顯示在可視化界面上。

圖1 系統架構圖

1.3 城市消防聯網遠程監控系統的云計算平臺高實時性的體現

城市消防聯網遠程監控系統是一個需要實時反饋各個消防單位狀態信息的系統，需要不斷的獲取消防單位數據服務器上的最新數據。因此數據采集的速度也直接影響系統能否做到高實時性，各消防單位采用規定好數據格式的log文件存儲實時數據，并存儲在數據服務器中，消防中心只讀消防單位log文件上的新增數據，大大減小數據傳輸量。消防單位中的log文件按照按天存放，有效控制單個log文件的大小。這樣，相較于傳統數據庫檢索方式，數據采集做到了高效穩定。

1.4 城市消防聯網遠程監控系統的云計算平臺高可靠性的體現

消防中心的服務器集群通過一個主服務器master服務器控制，集群中的服務器每隔一段時間就會發送一條狀態信息給master服務器，一旦服務器集群中的某個服務器出現故障，master服務器收不到該服務器的心跳信息，會將該服務器所處理的工作及時分配給別的服務器處理。并且master節點也有一個備用服務器，一旦master服務器宕機，備用服務器會立即投入使用。同時消防監控中心和消防單位之間也有一套心跳檢測機制，消防單位中的服務器設備每隔一段時間就會發送一條狀態信息給消防中心，如果狀態信息接收不正常，則通知相應的監控單位對相關服務器設備進行及時地檢修，保證平臺的穩定運行。這種機制也保證了消防監控中心和消防單位之間保持可靠接連。

1.5 城市消防聯網遠程監控系統的云計算平臺高可擴展性的體現

系統中的消防單位的識別信息都存在消防中心的master服務器的配置表中，對消防單位的編輯，只需修改配置表消防單位的信息。同時，如果發現集群中服務器的平均負載率過高(超過80%)，那么可以很方便的向集群中添加新的服務器，同樣是在master服務器的配置表中添加新并入服務器的信息。

本文的系統平臺進過實驗，為城市消防聯網遠程監控系統提供了經濟，穩定，高效，可靠，和方便擴展的云計算平臺。

2 城市消防聯網遠程監控系統系統中數據的實時處理

本文針對消防系統中海量數據的實時采集和處理需求提出基于flume、kafka、Strom集群的城市消防聯網實時監控系統。采用模塊設計，每個組件完成自己特定的業務功能。本章主要介紹消防監控系統的組成部分、體系架構設計、和系統客戶端結果展示等內容。

2.1 系統的組成部分

Flume是一個高可用的，高可靠的，分布式的海量日志采集、聚合和傳輸的系統。本系統通過部署在消防單位上數據服務器上的flume_agent，實時的采集log文件中新增的信息數據。

kafka是一種高吞吐量的分布式發布訂閱消息系統。因為需要在flume采集信息數據和storm處理數據之間建立緩沖中間件，所以使用kafka接受flume采集的數據，同時將數據以消息隊列的形式輸送給storm。

Storm 是一個開源的、大數據處理系統，旨在用于分布式實時處理且與語言無關。其使用核心在于拓撲結構的設計，我們需要把kafka中的消息數據傳到到storm里，再照著設計好的拓撲結構給消防中心服務器集群中的機器分發數據進行處理，因此根據實際需求確定好機器數量是整個系統處理效率的關鍵。 Storm運行在一個分布式的集群架構上，Storm集群由一個主節點和多個工作節點組成。

2.2 系統體系架構設計

城市消防聯網遠程監控系統采用四層架構：數據采集層、數據接入層、數據處理層和數據輸出層。如圖2所示。

圖2 Storm下的體系架構示意圖

2.2.1 數據采集層

使用Flume實現分布式日志收集系統進行日志采集：log文件中海量數據的高效收集是本系統的基石，本系統使用的Flume來進行日志采集。Flume本身采用了分層架構：分別為agent，collector和storage，Agent負責日志收集工作，Collector層負責接收Agent層發送的日志，并且將日志根據路由規則寫到相應的Store層中，Store層負責提供永久或者臨時的日志存儲服務，或者將日志流導向其它服務器。

Flume具有的:(1)高可靠性; (2)易擴展及管理性 等優點是本系統采用其作為日志采集工具的主要原因。

其中(1)高可靠性體現在：當系統中消防單位服務器節點出現故障時(例如agentA出現了故障)，其數據服務器上的日志文件能夠被傳送到備用數據服務器節點上(BackAgent1)而不會丟失。同時，Flume提供了3種級別的可靠性保障，從強到弱依次分別為：end-to-end(收到數據agent首先將event寫到磁盤上，當數據傳送成功后，再刪除；如果數據發送失敗，可以重新發送。)，Store on failure(這也是本系統采用的策略，當數據接收方服務器發生崩潰時，將數據寫到本地，待恢復后，繼續發送)，Best effort(數據發送到接收方后，不會進行確認)。

(2)易擴展及管理性：Flume采用了三層架構，分別為agent，collector和storage，每一層均可以水平擴展。其中Agent層中，每個監控單位的日志服務器部署一個進程，負責對監控單位的日志收集工作；Collector層部署在消防中心服務器上，負責接收Agent層發送的日志，并且將日志根據路由規則寫到相應的Store層中；Store層負責將日志存儲在本系統的下一個模塊：kafka。

此外，本系統在Agent層向Collector層放送數據的時候使用負載均衡策略，將所有的日志均衡地發到所有的Collector上，達到負載均衡的目標，避免了單點故障的問題。

圖3 flume系統架構圖

2.2.2 數據接入層

由于flume采集日志數據的速度和storm處理數據的速度是不同的，所以本系統使用kafka作為中間的一個緩沖模塊。kafka能夠實時的收集反饋信息，并能夠支撐較大的數據量，通過磁盤數據結構提供消息的持久化，這種結構對于即使數據達到TB+級別的消息，存儲也能夠保持長時間的穩定，且具備良好的容錯能力。本系統使用kafka統一接收flume采集的數據，將消息提交給系統的下一個模塊，storm集群進行數據的處理。

2.2.3 數據處理層

storm架構組成如圖4所示。其中有3種重要節點：Nimbus節點、Supervisor節點和數據庫節點。Nimbus就相當于Hadoop中的“JobTracker”，是用戶和Storm系統溝通的關鍵點，Nimbus主要負責發布和協調拓撲中的任務執行。Nimbus是無狀態的，如果Nimbus服務失敗了，工作節點仍然可以持續地進一步工作，只是無法進行任務調度，直到Nimbus重新復活。Supervisor運行在各個Storm節點上，它接受Nimbus分配的任務并根據任務啟動和停止屬于自己管理的工作進程。Supervisor也是快速失敗的，Nimbus和Supervisor所有的狀態都存儲在Zookeeper中，這樣設計是Storm具有恢復能力的關鍵。對分析后的結果持久化，可以用MySQL存儲，同時MySQL采用主從復制架構，來避免數據的丟失。

圖4 消防預警系統Storm架構組成示意圖

Storm數據處理架構由流元組組成的拓撲，有著明確的分層架構：Spout和Bolt，Spout是拓撲中數據的來源，它會從外部數據源中讀取數據，并發送給處理元組的Bolt，并通過拓撲中的系統及組件Acker追蹤從Spout中流出的元組的處理路徑，如果在用戶設置的最大超時時間內這些元組沒有被完全處理，那么Acker就會告知Spout該消息處理失敗了，相反會告知Spout消息處理成功了。這個Spout代表著整個元組的完全處理，如圖5所示。因此可以說Storm記錄容錯原理保障了消息的可靠處理。

表1 數據處理量和處理時間表

從一個提供數據源的Spout或Bolt流到處理元組的Bolt有很多種方式，由流分組機制所定義，主要有：隨機分組、字段分組、直接分組、全局分組等。本系統根據實際需求采用先字段分組后隨機分組策略。把數據流先按照消息類型字段(消息類型分為火警，故障，聯動，反饋等)分組，發送到不同的Bolt中，該Bolt再把接收到的某類型的數據流按照隨機分組方式分發元組到其它的Bolt任務中，對該數據進行計數等其他操作。最后再由整合Bolt計算統計結果。

消防單位中的各類探測設備和其他電子設備，遇到設定情況時會發送自己的警報數據，同時，每時每刻也在發送自己自檢數據，對這些數據的處理實時性要求比較高，而數據處理組件Bolt主要按字段分組過濾掉海量數據中正常的狀態數據，處理實時性要求比較高的報警數據，例如火警，故障，聯動等信息，然后將報警數據發送到web客戶端顯示以及批量插入數據庫中。插入完成后，追蹤組件Acker通知storm框架任務已經執行完成。

圖5 Bolt組件處理過程

2.2.4 數據輸出層

對于storm處理后的數據則需要顯示在web客戶端頁面上以及將數據存儲到數據庫中，用于后期的對這些數據的統計查詢。這里可以使用mysql分布式數據庫系統來存儲這些數據。這里不再詳述。

3 客戶端結果顯示

使用express+socket.io技術實現web客戶端更新數據：Storm處理完成數據處理后，通過express+socket.io技術實現更新客戶端信息界面，這里對于這個技術細節不再詳細贅述。消防監控中心對頁面上顯示的問題進行處理。有效的監管各個消防中心的運行。

4 性能測試

本系統主要提升了城市消防聯網遠程監控系統在大數據處理過程中的實時性和可靠性，下面分別從這兩個方面驗證系統性能。

4.1 實時性性能評估

測試系統采用4臺主機組建storm集群，CPU 4×3.3 G,4G內存，千兆以太網卡，系統環境：Java1.6+Kafka+storm-0.9.1+flume1.5+MySQL5.1.69，集群中1個為master服務器，測試情況下采用10不同大小的log文件，分別包含為5萬到50萬行數據來模擬不同的數據量。系統每次處理一個文件。測試過程中實時記錄集群中master服務器的CPU 、 I/O 和 內存占用情況以及其它3個節點服務器的硬件資源使用情況以及完成處理過程所用的時間。

圖6 數據處理量

如表1和圖6所示，從5 W到50 W行的數據，數據量線性增加，系統處理數據的所用時間呈緩慢增加。說明系統在處理大數據量時所用時間是可控的，實時性可以得到保證。同時我們看到無論是master服務器還是其它節點服務器的資源占用率都維持在正常水平。從這次實驗我們看到數據處理時間還是不夠理想，這跟實驗中服務器的硬件水平和數量有很大關系。實際使用中，根據需要增加服務器的數量和提高硬件水平可以有效縮短處理時間，提高實時性。

4.2 可靠性性能評估

在這個實驗中，我們預分配8臺主機。在拓撲中每個組件的初始數目如下表2。

表2 組件的初始數目

工作進程的數據設置為30，并且一直保持在30。我們在8臺機器上同時運行拓撲任務。然后，我們等待了15分鐘之后，關掉了其中的1臺機器，之后每隔15分鐘就關掉其中1臺機器。實驗的設置如表3所示。

表3 實驗實時配置展示表

最后，我們監視每分鐘處理的元組的數目來表示整個架構的吞吐量，同時也監視著一個元組的平均處理延遲。吞吐量時根據每分鐘被追蹤的元組的數據來衡量的。實驗的結果如表4所示。

表4 實驗吞吐量和延遲響應數據表

正如圖7所示，每當我們移除一組機器時，都會有一個臨時的下降尖峰，但是隨后很快地又恢復回來。同時，我們主要到，吞吐量每十五分鐘下降一次，這意味著同樣的拓撲被更少的機器處理。也正如圖所示，吞吐量也很快地穩定回來。

圖7 吞吐量測量圖

如圖8所示實驗的延遲圖，每關掉一組機器后，平均的響應延遲也增加了。但是我們同時也注意到了，在前幾個15分鐘內，延遲的時間是很短的，但是在最后兩個15分鐘內，延遲就相對來說有點大，但是，系統也都能夠很快地穩定處理任務。

圖8 延遲測量圖

總得來說，正如在這個實驗中所示，Storm對于機器故障具有很好的恢復能力。并且當面對機器故障時，也能夠有效地穩定系統的性能。

5 結言

論文主要是研究消防監控中心對各個消防單位的數據進行實時采集和處理，對各個消防單位進行實時監控、警情及時發現、故障及時反饋。通過分析消防單位和消防主管部門的相應的需求和實際狀況，我們發現利用基于云計算平臺的storm集群系統能夠有效地解決消防數據處理延遲以及系統可靠性問題。同時針對消防單位服務器或者是網絡出現故障時能夠及時發現，我們采用了心跳檢測機制來保障監控中心實時地了解消防消防單位的狀況，也保證了消防數據的準確無誤。同時這套系統也有很高的經濟性，消防單位的系統只需做少量修改，同時消防監控中心根據實際需求靈活配置服務器數量，

storm架構在解決大量數據流的實時處理方面具有很好的性能，也是以后未來工作的研究重點，在之后的工作中，我們會更多地改進可視化的方法，提升相關部分的可靠性，也會更多地考慮實時處理和批處理的相結合等問題。

[1] Arvind Arasu, Brian Babcock, Shivnath Babu, Mayur Datar, Keith Ito, Rajeev Motwani, Itaru Nishizawa, Utkarsh Srivastava, Dilys Thomas, Rohit Varma, Jennifer Widom: STREAM: The Stanford Stream Data Manager[J]. IEEE Data Eng. Bull. 2003, 26(1): 19-26.

[2] Minos N. Garofalakis, Johannes Gehrke: Querying and Mining Data Streams: You Only Get One Look[J]. VLDB 2002.

[3] Daniel J. Abadi, Yanif Ahmad, Magdalena Balazinska, Ugur Cetintemel, Mitch Cherniack, Jeong-Hyon Hwang, Wolfgang Lindner, Anurag Maskey, Alex Rasin, Esther Ryvkina, Nesime Tatbul, Ying Xing, Stanley B. Zdonik: The Design of the Borealis Stream Processing Engine[J]. CIDR 2005: 277-289.

[4] Apache Kafka, A high-throughput distributed messaging system. 2013[EB/OL]. http://kafka.apache.org/design.html.

[5] Storm. 2013[EB/OL]. http://storm.project.net/.

[6] S4 Distributed stream computing platform[EB/OL]. http://incubator.apache.org/s4/.

[7] Borthakur D, Sarma JS, Gray J, Muthukkaruppan K, Spigeglberg N, Kuang HR, Ranganathan K, Molkov D, Mennon A, Rash S, Schmidt R, Aiyer A Apache Hadoop goes realtime at Facebook[A]. Proc. Of the ACM SIGMOD Int”1 Conf. on management of Data(SIGMOD 2011 and PODS 2011)[C]. Athens: ACMPress,2011. 1071-1080.[doi:10.1145/1989323.1989438].

[8] Tyler Akidau, Alex Balikov, Kaya Bekiroglu, Slava Chernyak, Josh Haberman, Reuven Lax, Sam McVeety, Daniel Mills, Paul Nordstrom, Sam Whittle: MillWheel: Fault-Tolerant Stream Processing at Internet Scale[J]. PVLDB, 2013, 6(11): 1033-1044.

[9] Spark Streaming[EB/OL]. http://spark.incubator.apache.org/docs/latest/streaming-programming-guide.html.

[10] Rajagopal Ananthanarayanan, Venkatesh Basker, Sumit Das, Ashish Gupta, Haifeng Jiang, Tianhao Qiu, Alexey Reznichenko, Deomid Ryabkov, Manpreet Singh, Shivakumar Venkataraman: Photon: fault-tolerant and scalable joining of continuous data streams[A]. SIGMOD Conference 2013[C]. 577-588.

[11] Mohamed H. Ali, Badrish Chandramouli, Jonathan Goldstein, Roman Schindlauer: The extensibility framework in Microsoft StreamInsight[J]. ICDE 2011: 1242-1253.

[12] Sankar Subramanian, Srikanth Bellamkonda, Hua-Gang Li, Vince Liang, Lei Sheng, Wayne Smith, James Terry, TsaeFeng Yu, Andrew Witkowski: Continuous Queries in Oracle[J].VLDB 2007: 1173-1184.

[13] IBM Infosphere Streams[EB/OL]. http://www03.ibm.com/software/products/en/infosphere-streams/.

City Fire Remote Minitor Network System Based on Storm for Real-time Data Processing Application

Yang Susu

(College of Information Engineering, Capital Normal University, Beijing 100048, China)

In the city fire control network system, the data of real-time information has become larger and larger. The traditional fire control system cannot deal with the problem of real time data with high efficiency. We analyze the requirements and performance of the open source Storm framework, and implement the improvement on the technical architecture and the characteristics of the fire system. We put forward a set of system architecture for real-time processing of data in high real-time and high scalability of the fire control network monitoring center. At the same time, the construction of the cloud computing platform, using the heartbeat detection mechanism to ensure the real-time performance of the monitoring unit. The research shows that, based on cloud computing and Storm platform, the architecture is fully applicable to the fire control system, which has the characteristics of high efficiency, high reliability, high performance and so on.

Storm; fire protection pre-warning; big data real-time processing; cloud computing

2016-01-12；

2016-02-24。

國家科技支撐計劃項目(2013BAH19F01)。

楊素素(1990-)，女，碩士研究生，主要從事云計算與數據實時處理系統及應用方向的研究。

1671-4598(2017)03-0055-05

10.16526/j.cnki.11-4762/tp.2017.03.016

TP399

A