管道完整性云服務平臺分布式實時數據采集計算系統的設計與實現

毛愛芹 趙淑媛 劉偉

摘? 要： 本文對基于管道完整性云服務平臺分布式實時數據采集計算系統的設計方案、技術選型、關鍵技術等做了詳細的研究，最終實現了基于ApacheZookeeper、ApacheKafka、ApacheStorm以及Redis的實時數據采集系統，解決了管道完整性云服務平臺在面對海量數據并發以及復雜業務計算等場景下的吞吐量瓶頸，為基于大數據的管道完整性業務的深度數據挖掘、離線/在線數據分析提供了可靠的數據處理框架。

關鍵詞：管道完整性 云服務平臺 實時數據 Zookeeper Kafka Storm Redis

【中圖分類號】TP? ? ? ?【文獻標識碼】A

1 引言

管道完整性管理是一個系統的、綜合性的管理體系，是通過一定技術和管理體系對可能存在的對管道安全造成影響的危險點進行識別、預防及改進，從而避免事故的發生，始終保持管道在整個生命周期內處于完整、安全、良好的狀態[2]。

云GIS，是指以云計算為架構的網絡化、虛擬化、分布式、按需獲取的地理信息系統。實質是將GIS的平臺、軟件和數據方便、快捷、高效地部署到云基礎設施之上，能夠以彈性的、按需獲取的方式提供最廣泛的WebGIS服務。云GIS連接智能移動終端，用戶可在任意位置獲取超大規模的GIS功能。云GIS將并行計算與GIS結合，引入并行數據庫，組成集群系統，利用并行存儲、查詢、檢索，處理等技術，有效提升傳統GIS對海量空間數據的高效存儲與處理。專業級云GIS平臺建設內容包含硬件支撐環境搭建、系列標準建設、云資源池建設、數據服務體系建設、系列軟件研發、應用系統建設[1]。

由于管道完整性管理的海量數據及空間可視化的需求，地理信息系統技術在管道完整性管理中的作用越來越大[4]。將云GIS引入管道完整性管理中，利用云服務高并發、分布式計算、分布式存儲等特點實現對海量業務數據的高效處理，以此為基礎實現對海量數據的深度挖掘，獲取管道完整性管理中重要的數據和指標，使得在管道生命周期內每時每刻產生的業務數據能夠真正為決策者提供有價值的情報。

2 關鍵技術

管道完整性云服務平臺的分布式實時數據采集平臺旨在解決海量的高并發數據流，實現低延遲響應處理，保證數據流的強時序性，實現復雜空間計算，并且分布式系統需要與集中式系統完全不同的系統架構。因網絡會損失信息，所以需要軟件進行恢復。當網絡出現過載時，也必須容錯恢復[5]。

2.1 高吞吐量消息引擎

管道完整性云服務平臺面向長輸管道完整性管理的各個環節，包括：管道巡檢、管道應急、施工期管理、移動辦公等多個領域，各個業務板塊產生大量的實時數據，因此如何響應并處理這些海量數據成為云服務平臺急需解決的問題。

Apache Kafka是一個開源消息系統項目，由Scala寫成。該項目的目標是為處理實時數據提供一個統一、高通量、低等待的平臺。它通過一種獨一無二的設計提供了一個消息系統的功能， 解決了大規模分布式應用中數據傳遞的高效與穩定的問題，解決了從集中式向分布式采集架構過渡的難題[6]。

2.2 分布式流數據計算

地理空間數據的處理結果需要及時反饋給用戶，例如對每次移動智能終端上傳的巡檢點數據，系統需要及時計算其與管道的距離，當超過某一閾值時，移動終端將及時收到服務器端的警報信息。面對海量的數據計算需求，傳統的單一服務器、單一計算節點的方式無法保證計算的低延時及強時序性，因此如何實現實時地理數據的預處理成為云服務平臺的重要問題。批量計算和流式計算適用于不同的場景，對于數據實時性要求不高，且數據越多，計算結果越準確的場合批量計算更合適，但對于實時性要求高，只需對最近部分進行處理分析的場合，流式計算有著明顯優勢[7]。

Storm是一個分布式的，可靠的，容錯的數據流處理框架。Storm集群的輸入流由一個被稱作spout的組件管理，spout把數據傳遞給bolt， bolt要么把數據保存到某種存儲器，要么把數據傳遞給其它的bolt。一個Storm集群就是在一連串的bolt之間轉換spout傳過來的數據。

2.3 分布式數據緩存

經過流式計算框架處理的地理空間數據可以持久化到各種物理存儲單元中，比如傳統的關系型數據庫（Oracle、Sql Server等）以及NoSql數據庫（MongoDB、Apache Hbase等），但是由于云服務平臺的海量數據并發造成對其的預處理結果也具有高并發的特點，因此對其的持久化操作不應該直接訪問數據庫造成大量的IO讀寫開銷。針對這一問題，需要設計一個分布式基于內存的數據緩存框架。Key-Value存儲以其高擴展性和強一致性在如今熱門的大數據時代中扮演著重要角色。而實現其擴展性重要的關鍵技術在于數據遷移機制的完善，數據遷移機制同時直接影響集群節點的負載均衡。

3 分布式實時數據采集平臺設計

圖1所示，分布式實時數據采集平臺由客戶端、服務協調組件集群、消息隊列集群、流計算集群、消息緩存集群和數據庫集群6個部分組成。

3.1 客戶端

客戶端作為業務數據的生產者主要由手持端設備和web系統組成，其產生的實時位置數據、報警數據、SCADA監控數據通過網絡傳輸至分布式數據平臺，在傳輸過程中，使用數字簽名等技術對傳輸內容進行加密，保證傳輸安全。

3.2 服務協調組件

由于分布式數據采集平臺由多臺服務器組成，為保證各服務節點之前的高可用（HA），平臺采用Zookeeper作為分布式協調組件，記錄各服務節點的元數據，實現服務節點間的心跳檢測，確保整個平臺在生產環境下的健壯性，防止由于網絡抖動、服務器宕機造成平臺無法正常對外提供服務。

3.3 消息隊列集群

消息隊列集群負責接收客戶端產生的多種業務數據，平臺采用Apache Kafka作為消息總線，Kafka是一個吞吐量極高的分布式消息系統，其整體設計是典型的發布與訂閱模式系統。由于kafka的高吞吐量，可以確保來自客戶端的每條數據被正確處理。由于業務數據的多樣化，平臺針對不同種類的業務數據建立不同的topic分類，并采用自定義分區策略（時間熱度分區算法、哈希散列值分區等）將數據寫入topic的不同partition中，同時為保證數據的安全性對每一個topic的partition建立多個副本，存儲結構如下圖所示：

3.4 流計算集群

實時流計算主要應用于兩個場景：持續計算，一旦任務啟動除非人為終止否則會一直處理到達數據;支持分布式計算，對大數據采用并行計算以降低單個處理器任務量[9]。由客戶端寫入kafka的業務數據，需要被實時高效、低延遲地處理并及時反饋給客戶端，為滿足這一要求，平臺采用Apache Strom作為流式計算服務，以管道完整性云服務平臺的智能管道巡檢系統為例，Strom對不同巡檢人員上傳的巡檢點位置數據進行快速處理，計算其超時/超速/超距等報警信息，其處理流程如下圖所示：

1）KafkaSpout作為消費者訂閱Kafka Topic，從Broker的partition中拉取消息。

2）GlobeBolt對原始數據進行預處理，并采用Shuffle Grouping作為數據分組策略，以保證Strom的并發處理效率，消除數據傾斜。

3）預處理后的數據同時發射至SocketBolt和GroupCalcuateBolt進行處理，SocketBolt負責將處理后的數據推送至Socket服務器保證客戶端隨時可以接收到實時數據，GroupCalcuateBolt負責對數據進行分組和業務計算，計算結果同樣推至至Socket服務器，該節點采用FieldsGrouping策略，以保證具有相同key值的數據總是被同一bolt實例處理，實現數據的連續處理。

4）RedisBolt工作節點負責將GroupCalcuateBolt的計算結果寫入Redis緩存，并在時間窗口內將結果進行持久化操作。

3.5 消息緩存集群

由于Strom時刻都在處理業務數據，因此采用直接將計算結果做持久化處理將造成大量的數據庫IO操作，增加服務器開銷，因此平臺將計算結果以鍵值對的形式寫入Redis內存數據庫，再使用定時器任務批量將Strom計算結果寫入MongoDB物理數據庫。在高負載的情況下，添加更多的節點，可以保證服務器性能。MongoDB旨在為Web應用提供可擴展的高性能數據存儲解決方案[10]。

為保證Redis的高可用性，需要在發生主節點服務器故障時能夠迅速做到主從切換，將slave節點的角色變更為master，因此在集群主節點所在的服務器中開啟Redis哨兵進程，以監控主節點狀態。

4 系統在管道完整性云服務平臺中的應用

基于Kafka/Storm的分布式實時數據采集計算系統實現了巡檢軌跡數據采集、巡檢三超報警（超速、超時、超距）計算、SCADA監控數據采集以及系統日志采集等功能，實現了海量業務數據的并發請求處理，結合ACK技術，保證了每一條業務數據在數據流中能夠被正確及時地處理，由于系統實時地將熱點數據存入Redis數據庫，用戶對熱點數據的請求經過Web容器的攔截全部發往Redis，命中后返回給客戶端，極大地提升了請求響應地速度。熱點數據過期后自動持久化到MongoDB數據庫，為進一步的離線分析生產基礎數據。

系統目前已經計入全面測試階段，下一步將在各油氣儲運單位推廣應用。分布式實時數據采集計算系統將解決單線程阻塞式地處理方式造成的野外巡檢軌跡點上傳后不定時地丟失、SCADA監測數據高頻率發送造成的web容器崩潰以及用戶關心的熱點數據由于數據量過大造成的前端頁面響應時間長等問題，將極大的提升管道完整性云服務平臺的健壯性和數據吞吐量，發揮大數據技術在管道安全方面的巨大作用。

5 總結與展望

5.1 系統特點

1）采用RestfulAPI結合Apache Kafka消息總線技術實現對客戶端上傳的海量數據采集功能，保證了各業務數據的完整性，并通過設置kafka的相關配置實現了集群的高可用。

2）采用Apache Strom做為流式計算框架，使得原本高延遲的地理空間數據計算可以在極短的時間窗口內完成，并且在計算工作節點中采用滑動窗口算法，保證了數據完整性和準確性。

3）系統采用Redis集群做為緩存解決方案，保證了內存數據的可靠性，減少了Storm直接訪問數據庫造成的額外網絡傳輸開銷和內存讀寫消耗。

4）系統采用HBase做為海量實時數據的存儲解決方案，基于HBase的列式存儲方案，極大的減少了IO操作，同時由于HBase是運行在Hadoop上的分布式可擴展的NoSql大數據倉庫，因此為后期基于MapReduce的離線分析提供了便利的數據存儲方式。

5.2 展望

管道完整性云服務平臺分布式實時數據采集系統基于Apache Kafka、Strom、Redis等流行開源技術實現了海量業務數據的收集、計算、存儲功能，該系統在長輸管道、城市燃氣、智慧城市等方面都具有廣泛的應用前景。在下一步的工作中，將針對管道巡檢關鍵點到達判定、人員車輛實時位置結合地理圍欄技術的監控等業務做進一步的研究和開發，同時對管道完整性的智能化進行深入研究，為日常管道安全管理工作的開展提供了完整、準確、及時的技術和決策支持。

參考文獻：

[1] 陶旸，劉昱君.面向行業深度耦合的專業級云GIS服務平臺設計[J] .測繪與空間地理信息，2018：15-16.

[2] 帥義，帥健，馮燦.基于B/S架構的管道完整性管理系統[J] .油氣儲運，2017，36 （04） ：388.

[3] 王宇寧.基于Android的長輸管道巡線系統的研發[D].西安：西安石油大學，2015：5-11.

[4] 徐杰，曾晟 劉亮等.基于APDM的管道完整性數據管理研究[J].天然氣與石油，2016，34（05）：76.

[5]陳天偉，彭凌西 基于ZooKeeper的一種分布式系統架構設計與實現[J].通信技術，2018，51 （01）：87-90.

[6] 金雙喜，李永 基于Kafka消息隊列的新一代分布式電量采集方法研究[J].電網分析與研究，2018，46 （02）：78.

[7] 周明陽 閆超 郭良浩等基于Storm的分布式實時信號處理系統[J].計算機應用，2017，37 （z1）：69.

[8]? 韋立，陳姍姍 基于Redis單位最大效益自適應遷移策略研究[J].計算機技術與發展， 2018，28 （10） ：54-58.

[9]? 李一辰，李緒志，閻鎮 實時流計算在航天地面數據處理系統中的應用[J].微電子學與計算機， 2014，31（09） ：15-19.

[10]? 戴傳飛，馬明棟 MongoDB分頁技術優化研究[J].計算機技術與發展，2018，28 （06） ：98-101.

[11]孫大為，張廣艷，鄭緯民 大數據流式計算： 關鍵技術及系統實例[J].軟件學報，2014，25（ 4） ： 839-862.

補充說明：本文沒有基金項目資助。

作者簡介：毛愛芹（1983 — ），女，安徽省淮北市，高級工程師，碩士學歷（2008年畢業于中國礦業大學 地圖制圖學與地理信息工程專業）。主要從事地理信息系統建設、測繪、管道完整性管理應用等工作。

地址：陜西省西安市長安區 韋曲街道 神舟四路216號中煤航測遙感集團 郵編：710199

電話：029-89692117? 15802980764

E-Mail：94209732@qq.com

創新點名稱：管道完整性云服務平臺分布式實時數據采集計算系統

創新點內容：基于分布式系統架構，結合高效的消息隊列引擎和高吞吐量的流式數據計算引擎實現了地理實時數據的在線計算處理，處理結果使用Redis緩存集群，采用AppendOnly持久化方案，確保處理結果的完整性。系統能夠實現各種實時數據（巡檢軌跡點、SCADA監控數據、人員/車輛/設備位置等）的收集，并結合完整性業務場景對海量數據進行低延遲計算。該系統充分結合管道巡檢、管道應急、數據監控管理的需求，適合國內管道完整性數據采集的推廣，有較強的應用價值。