彭大為，史惠存

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

0 引言

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

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

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

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

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

1 Rocket MQ

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

圖1 RoketMQ結構體

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

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

Roket MQ為了實現高吞吐量、高并發(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

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

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

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

圖3 Dstream

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

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

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

窄依賴 窄·寬依賴

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

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

3 總體架構

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

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

圖5 總統(tǒng)架構

4 Spark Streaming調優(yōu)

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

4.1 接收和處理的并行度

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

4.2 數據序列化

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

4.3 批處理時間間隔設置

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

5 結論

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

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

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