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

彭大為，史惠存

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

0 引言

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

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

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

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

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

1 Rocket MQ

RocketMQ是一個(gè)由阿里巴巴開(kāi)源的消息中間件，在設(shè)計(jì)上借鑒了Kafka。2012年開(kāi)源，2017年成為apache頂級(jí)項(xiàng)目。RoketMQ結(jié)構(gòu)體如圖1所示。

圖1 RoketMQ結(jié)構(gòu)體

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

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

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

圖2 Topic Broker Queue關(guān)系

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

2 Spark Streaming

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

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

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

圖3 Dstream

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

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

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

窄依賴(lài) 窄·寬依賴(lài)

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

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

3 總體架構(gòu)

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

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

圖5 總統(tǒng)架構(gòu)

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

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

4.1 接收和處理的并行度

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

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

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

4.3 批處理時(shí)間間隔設(shè)置

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

5 結(jié)論

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

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

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