水利自動化實時流式大數(shù)據(jù)的處理研究

彭大為，史惠存

(1.安徽省水利水電勘測設計研究總院有限公司，230088，合肥；2.江蘇省建筑工程質(zhì)量檢測中心有限公司，210028，南京)

0 引言

根據(jù)《促進大數(shù)據(jù)發(fā)展行動綱要》和《全國水利信息化發(fā)展“十三五”規(guī)劃》提出加快推動數(shù)據(jù)資源共享開放和開發(fā)應用， 實施國家大數(shù)據(jù)戰(zhàn)略?！蛾P于推進水利大數(shù)據(jù)發(fā)展的指導意見》指出應用大數(shù)據(jù)處理技術處理多源、海量、動態(tài)、持續(xù)增加的水利數(shù)據(jù)，以滿足智慧水利的建設要求。在《水利部關于印發(fā)加快推進智慧水利的指導意見和智慧水利總體方案的通知》中強調(diào)基于物聯(lián)網(wǎng)建設空天地一體化數(shù)據(jù)感知網(wǎng)絡，采集水利相關數(shù)據(jù)，為水利大腦提供數(shù)據(jù)支撐。中國水利企業(yè)協(xié)會智慧水利分會在智慧水利與河湖長制高峰論壇暨2019智慧水利分會年中再一次圍繞云計算、大數(shù)據(jù)、物聯(lián)網(wǎng)等技術實現(xiàn)由數(shù)字水利向智慧水利轉(zhuǎn)變。水利大數(shù)據(jù)是實現(xiàn)智慧水利的前提和基礎。

隨著水利行業(yè)智能傳感器的發(fā)展和大規(guī)模的應用以及物聯(lián)網(wǎng)技術的逐漸成熟，水利行業(yè)數(shù)據(jù)采集能力不斷提升，形成水利大數(shù)據(jù)。水利大數(shù)據(jù)是由水利業(yè)務數(shù)據(jù)、水利相關行業(yè)和領域數(shù)據(jù)構成。龔琪慧[1]等提出水利大數(shù)據(jù)具有數(shù)據(jù)量大、數(shù)據(jù)來源形式多樣、數(shù)據(jù)持續(xù)增長、數(shù)據(jù)價值高和數(shù)據(jù)有實時性或準實時要求等特征。陳蓓青[2]等認為水利大數(shù)據(jù)具有數(shù)據(jù)量大、數(shù)據(jù)類型復雜和計算過程復雜耗時等特點。

水利數(shù)據(jù)分為結構化數(shù)據(jù)、非結構化、半結構化數(shù)據(jù)，結構化數(shù)據(jù)是關系型數(shù)據(jù)，有數(shù)據(jù)結構規(guī)范；非結構化數(shù)據(jù)是數(shù)據(jù)結構不完整不規(guī)則的數(shù)據(jù)形式；半結構化數(shù)據(jù)通常稱為自描述結構數(shù)據(jù)，是結構化數(shù)據(jù)的一種，包含相關標記，用來分隔語義元素以及對記錄和字段進行分層，如XML , GeoJson數(shù)據(jù)格式。為了高效處理水利大數(shù)據(jù)，通過采用分布式數(shù)據(jù)處理集群組建云環(huán)境，對不同結構的數(shù)據(jù)進行相應的處理。

傳統(tǒng)水利數(shù)據(jù)的處理是基于專業(yè)和部門的內(nèi)部數(shù)據(jù)進行抽樣分析，而水利大數(shù)據(jù)的處理是對跨部門、跨領域、多維度的海量數(shù)據(jù)進行整體分析處理。水利大數(shù)據(jù)分析一般是指利用分布式計算集群對數(shù)據(jù)進行處理、挖掘和可視化操作等， 主要有批處理和流處理2種方式[3]。MapReduce是目前較為常用的批處理引擎，但是在處理海量高頻實時數(shù)據(jù)時吞吐量較大，不能做到低延遲，不適合實時處理。Spark Streaming是Spark核心API的一個擴展，可以實現(xiàn)高吞吐量的、具備容錯機制的實時流數(shù)據(jù)的處理[4]。

本文主要闡述對水利自動化設備上報的實時流式大數(shù)據(jù)的處理方法，特別是通過PLC技術實時獲取自動化設備的運行參數(shù)，該數(shù)據(jù)上報頻率在毫秒級別，數(shù)據(jù)量大、數(shù)據(jù)源多、格式不同、數(shù)據(jù)持續(xù)等特點，使得對該類的數(shù)據(jù)處理一般傳統(tǒng)的數(shù)據(jù)處理模式已經(jīng)無法滿足，因此針對這種數(shù)據(jù)設計采用Spark Streaming技術進行處理，通過測試已證明本方法能夠?qū)λ詣踊瘜崟r流式大數(shù)據(jù)進行高效處理。

1 Rocket MQ

RocketMQ是一個由阿里巴巴開源的消息中間件，在設計上借鑒了Kafka。2012年開源，2017年成為apache頂級項目。RoketMQ結構體如圖1所示。

圖1 RoketMQ結構體

由Producer(生產(chǎn)者)、Topic(消息隊列)、Consumer(消費者)3部分組成。消息通過Topic進行傳遞。Topic存放的是消息的邏輯地址。Producer將消息發(fā)往具體的Topic。Consumer訂閱Topic，主動拉取或被動接受消息。

Topic是一個邏輯上的概念，每個邏輯隊列保存一部分消息數(shù)據(jù)，但是保存的消息數(shù)據(jù)實際上不是真正的消息數(shù)據(jù)，而是指向Commit log的消息索引。Top可采用集群的方式進行創(chuàng)建，也可通過單個Broker模式去創(chuàng)建，創(chuàng)建的每個Topic的角色相同。

Roket MQ為了實現(xiàn)高吞吐量、高并發(fā)，通常一個Topic被分配到多個代理中，每個代理包含多個Topic分區(qū)，每個Topic分區(qū)中保存相同類型的Queue。 Topic Broker Queue關系如圖2所示。

圖2 Topic Broker Queue關系

消息隊列作為RoketMQ高并發(fā)系統(tǒng)的核心組件，能夠幫助業(yè)務系統(tǒng)結構提升開發(fā)效率和系統(tǒng)穩(wěn)定性。

2 Spark Streaming

流式計算作為大數(shù)據(jù)處理領域的一種主要模型，當前主流的流計算框架由Twitter公司開發(fā)的Storm，Yahoo公司開發(fā)S4、微軟的Timestream以及UCBerkeley AMPLab開發(fā)的Spark Streaming等。

Spark Streaming實時流式大數(shù)據(jù)集群由多個工作節(jié)點組成，每個節(jié)點運行多個Spark Executor，在Spark Executor上運行相關業(yè)務處理程序。在此可以創(chuàng)建Maven應用來處理水利自動化上報的PLC數(shù)據(jù)，并將應用和相關Jar包進行完整打包，通過Spark-submit命令將Jar包提交到Spark集群節(jié)點中。提交應用的集群節(jié)點作為該應用的Driver節(jié)點，并從Cluster Manager中獲取資源，將根據(jù)時間窗口將實時流式數(shù)據(jù)進行分批形成彈性分布式數(shù)據(jù)集(RDD)，每次RDD的Action會產(chǎn)生一個新Job，每個Job包含多個Task，Cluster Manager機制根據(jù)集群資源分配情況動態(tài)地將Task分配到Worker Node中的Executor中處理。 Worker各節(jié)點最終處理結果匯集到Driver節(jié)點進行匯總輸出。

Spark Streaming是基于離散數(shù)據(jù)流Dstream，構建在Spark計算引擎之上的分布式流式計算框架，具有高吞吐量、具備容錯機制的實時流數(shù)據(jù)處理。Dstream是由時間上連續(xù)的彈性分布式數(shù)據(jù)集RDD序列組成，每個RDD包含一定時間間隔內(nèi)的數(shù)據(jù)流，是不可變的，可重算的數(shù)據(jù)集[5]。

圖3 Dstream

Spark將大數(shù)據(jù)切分后放入RDD作為Spark的基本數(shù)據(jù)結構，在RDD中進行數(shù)據(jù)處理操作，Spark根據(jù)操作調(diào)度集群資源進行計算。RDD的操作主要分為Transformation和Action 2種。Transformation操作表示將一個RDD通過一系列操作變成另一個RDD的過程。Transformation操作不會觸發(fā)真正的計算，僅建立RDD間的DAG有向無環(huán)圖。Action代表一次計算結束，不再生成新的RDD，并將結果返回給Driver程序。每個Action操作調(diào)用SparkContext的RunJob方法向集群提交請求。

通過有向無環(huán)圖(DAG)的Narrow窄依賴、Shuffle寬依賴實現(xiàn)Spark streaming的容錯。此種容錯主要是由于RDD中的數(shù)據(jù)是不可變的分布式彈性數(shù)據(jù)集，在集群處理過程中若RDD丟失可根據(jù)DAG找到其父RDD重新計算得到。

Narrow Dependency即父RDD與子RDD間的分區(qū)一對一，Map、Union操作為窄依賴；Shuffle Dependency即父RDD與子RDD間的對應關系不是一對一關系，Reducebykey操作為寬依賴；值得注意的是Join同屬寬、窄依賴如圖4所示。

窄依賴 窄·寬依賴

Spark Streaming支持從多種數(shù)據(jù)源獲取數(shù)據(jù)，如kafka、flume、zeroMQ、RoketMQ等。 Spark Streaming通過定義接口從RoketMQ消費者API中讀取數(shù)據(jù)，分配到Spark Streaming集群節(jié)點分區(qū)中，從而實現(xiàn)對實時流式大數(shù)據(jù)進行分布式處理[6]。

由于水利自動化PLC數(shù)據(jù)的高頻上報，數(shù)據(jù)持續(xù)到達，數(shù)據(jù)實時讀入和數(shù)據(jù)規(guī)模大特點的實時流式大數(shù)據(jù)的處理傳統(tǒng)的數(shù)據(jù)處理框架不能滿足需求，而Spark Streaming能夠很好地對實時流式大數(shù)據(jù)和離線大數(shù)據(jù)進行分布式處理[7]。

3 總體架構

基于RoketMQ和Spark Streaming框架設計了一種水利自動化實時流式數(shù)據(jù)分布式處理平臺?？傮w架構包括數(shù)據(jù)源、Netty服務器、RoketMQ、SparkStreaming和持久層，具體如圖5所示。

數(shù)據(jù)源是水利自動化設備通過將下位機中的采集和運行數(shù)據(jù)，通過PLC數(shù)據(jù)采集軟件將采集的數(shù)據(jù)由Netty服務器調(diào)用RoketMQ生產(chǎn)者接口寫入RoketMQ集群節(jié)點[8]。RoketMQ集群節(jié)點對持續(xù)實時上報的數(shù)據(jù)，根據(jù)不同告警類型的PLC數(shù)據(jù)寫入相應的Topic中，經(jīng)過相關業(yè)務處理后的數(shù)據(jù)去向分為3類，第1類是實時告警信息以主動的方式可通過消息系統(tǒng)SMS(如：移動、聯(lián)通、電信、網(wǎng)通、阿里云、微信、騰訊等)，將告警信息及時發(fā)送到相關負責人，使得告警能夠及時快速響應；第2類是將實時上報的PLC數(shù)據(jù)存入Elasticsearch數(shù)據(jù)庫中，由于Elasticsearch具有很強的數(shù)據(jù)檢索能力，系統(tǒng)在數(shù)據(jù)查詢時時效性很高；第3類是將復雜業(yè)務處理的數(shù)據(jù)傳遞到Spark Streaming進行集群運算，并將處理結果持久化到時序數(shù)據(jù)庫和關系數(shù)據(jù)庫中，該類數(shù)據(jù)庫可作為水利自動化上報的成果數(shù)據(jù)庫，為系統(tǒng)相關業(yè)務應用提供支撐。

圖5 總統(tǒng)架構

4 Spark Streaming調(diào)優(yōu)

將水利自動化實時流式大數(shù)據(jù)應用部署在集群之初，出現(xiàn)應用系統(tǒng)運行慢、占用資源多、不穩(wěn)定等問題，此時需要對集群進行調(diào)優(yōu)才能達到最佳性能。調(diào)優(yōu)是一個具體性很強的事情，不同的應用及場景優(yōu)化方式也不同，調(diào)優(yōu)并沒有統(tǒng)一的標準。

4.1 接收和處理的并行度

在分布式系統(tǒng)中增加接收和處理數(shù)據(jù)的并行度是提高整個系統(tǒng)性能的關鍵，為了提高數(shù)據(jù)接收和處理的并行度需要考慮數(shù)據(jù)分片的數(shù)量。每次Task僅能處理一個Partition,若Partiton個數(shù)過小導致每個Partition的數(shù)據(jù)量太大，導致內(nèi)存壓力過大，Executor的計算能力不能充分利用；若Partiton個數(shù)過多，導致分片過多，執(zhí)行效率低。

4.2 數(shù)據(jù)序列化

數(shù)據(jù)序列化在分布式數(shù)據(jù)處理過程中對集群性能的影響較為顯著。本文主要是采用Java的對象輸入流框架Object Output Stream Framework進行序列化，通過采用Java.io.Externalizable接口實現(xiàn)更加精細的控制序列化。

4.3 批處理時間間隔設置

水利自動化PLC上報的實時數(shù)據(jù)的頻率能否與集群的處理速度相對保持穩(wěn)定，直接影響數(shù)據(jù)處理的穩(wěn)定性。若上報的PLC數(shù)據(jù)的頻率過高于集群處理速度則將產(chǎn)生數(shù)據(jù)積壓，上報的PLC數(shù)據(jù)過低于集群處理速度造成算力浪費。合理的批處理時間間隔設置使得數(shù)據(jù)上報和處理保持一個相對穩(wěn)定的狀態(tài)，保證系統(tǒng)運行穩(wěn)定，否則產(chǎn)生延遲不斷增加，系統(tǒng)運行不穩(wěn)定。

5 結論

水利大數(shù)據(jù)處理技術是水利科學發(fā)展的必然趨勢，是大數(shù)據(jù)研究重要的領域。根據(jù)國家水利部2019年發(fā)布的智慧水利建設指導意見，要求構建水利一體化感知網(wǎng)絡構建水利數(shù)據(jù)采集平臺，該感知平臺基于物聯(lián)網(wǎng)技術獲取水利相關數(shù)據(jù)，為實現(xiàn)智慧水利提供數(shù)據(jù)支撐。

由于水利數(shù)據(jù)多源、異構、量大，特別是水利自動化實時上報的流式數(shù)據(jù)，在實際業(yè)務中往往需要對數(shù)據(jù)進行高頻率或超高頻率上報，數(shù)據(jù)不僅量大而且數(shù)據(jù)持續(xù)到達，同時要求處理響應低延遲，因此對水利大數(shù)據(jù)進行高效處理是實現(xiàn)智慧水利的核心技術。

根據(jù)實驗測試證明構建該分布式集群和采用實時流式大數(shù)據(jù)處理技術能夠?qū)崿F(xiàn)對水利自動化實時流式大數(shù)據(jù)進行高效處理。在該集群運行之初出現(xiàn)集群的數(shù)據(jù)處理效率很低、CUP和內(nèi)存的占用率較高的問題，通過對集群進行多次調(diào)優(yōu)后數(shù)據(jù)處理效率有較大提升，達到實驗設計目標。該水利自動化流式大數(shù)據(jù)處理可能存在一些不足，需要在項目實踐中進一步的改造、擴展、優(yōu)化以滿足實際項目建設需要。