彭大為，史惠存

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

0 引言

根據(jù)《促進(jìn)大數(shù)據(jù)發(fā)展行動(dòng)綱要》和《全國水利信息化發(fā)展“十三五”規(guī)劃》提出加快推動(dòng)數(shù)據(jù)資源共享開放和開發(fā)應(yīng)用， 實(shí)施國家大數(shù)據(jù)戰(zhàn)略?！蛾P(guān)于推進(jìn)水利大數(shù)據(jù)發(fā)展的指導(dǎo)意見》指出應(yīng)用大數(shù)據(jù)處理技術(shù)處理多源、海量、動(dòng)態(tài)、持續(xù)增加的水利數(shù)據(jù)，以滿足智慧水利的建設(shè)要求。在《水利部關(guān)于印發(fā)加快推進(jìn)智慧水利的指導(dǎo)意見和智慧水利總體方案的通知》中強(qiáng)調(diào)基于物聯(lián)網(wǎng)建設(shè)空天地一體化數(shù)據(jù)感知網(wǎng)絡(luò)，采集水利相關(guān)數(shù)據(jù)，為水利大腦提供數(shù)據(jù)支撐。中國水利企業(yè)協(xié)會(huì)智慧水利分會(huì)在智慧水利與河湖長制高峰論壇暨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ù)來源形式多樣、數(shù)據(jù)持續(xù)增長、數(shù)據(jù)價(jià)值高和數(shù)據(jù)有實(shí)時(shí)性或準(zhǔn)實(shí)時(shí)要求等特征。陳蓓青[2]等認(rèn)為水利大數(shù)據(jù)具有數(shù)據(jù)量大、數(shù)據(jù)類型復(fù)雜和計(jì)算過程復(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ù)通常稱為自描述結(jié)構(gòu)數(shù)據(jù)，是結(jié)構(gòu)化數(shù)據(jù)的一種，包含相關(guān)標(biāo)記，用來分隔語義元素以及對(duì)記錄和字段進(jìn)行分層，如XML , GeoJson數(shù)據(jù)格式。為了高效處理水利大數(shù)據(jù)，通過采用分布式數(shù)據(jù)處理集群組建云環(huán)境，對(duì)不同結(jié)構(gòu)的數(shù)據(jù)進(jìn)行相應(yīng)的處理。

傳統(tǒng)水利數(shù)據(jù)的處理是基于專業(yè)和部門的內(nèi)部數(shù)據(jù)進(jìn)行抽樣分析，而水利大數(shù)據(jù)的處理是對(duì)跨部門、跨領(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ù)的處理方法，特別是通過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ì)該類的數(shù)據(jù)處理一般傳統(tǒng)的數(shù)據(jù)處理模式已經(jīng)無法滿足，因此針對(duì)這種數(shù)據(jù)設(shè)計(jì)采用Spark Streaming技術(shù)進(jìn)行處理，通過測試已證明本方法能夠?qū)λ詣?dòng)化實(shí)時(shí)流式大數(shù)據(jù)進(jìn)行高效處理。

1 Rocket MQ

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

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

由Producer(生產(chǎn)者)、Topic(消息隊(duì)列)、Consumer(消費(fèi)者)3部分組成。消息通過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)建，也可通過單個(gè)Broker模式去創(chuàng)建，創(chuàng)建的每個(gè)Topic的角色相同。

Roket MQ為了實(shí)現(xiàn)高吞吐量、高并發(fā)，通常一個(gè)Topic被分配到多個(gè)代理中，每個(gè)代理包含多個(gè)Topic分區(qū)，每個(gè)Topic分區(qū)中保存相同類型的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)提升開發(fā)效率和系統(tǒng)穩(wěn)定性。

2 Spark Streaming

流式計(jì)算作為大數(shù)據(jù)處理領(lǐng)域的一種主要模型，當(dāng)前主流的流計(jì)算框架由Twitter公司開發(fā)的Storm，Yahoo公司開發(fā)S4、微軟的Timestream以及UCBerkeley AMPLab開發(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)用來處理水利自動(dòng)化上報(bào)的PLC數(shù)據(jù)，并將應(yīng)用和相關(guān)Jar包進(jìn)行完整打包，通過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的操作主要分為Transformation和Action 2種。Transformation操作表示將一個(gè)RDD通過一系列操作變成另一個(gè)RDD的過程。Transformation操作不會(huì)觸發(fā)真正的計(jì)算，僅建立RDD間的DAG有向無環(huán)圖。Action代表一次計(jì)算結(jié)束，不再生成新的RDD，并將結(jié)果返回給Driver程序。每個(gè)Action操作調(diào)用SparkContext的RunJob方法向集群提交請(qǐng)求。

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

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

窄依賴 窄·寬依賴

Spark Streaming支持從多種數(shù)據(jù)源獲取數(shù)據(jù)，如kafka、flume、zeroMQ、RoketMQ等。 Spark Streaming通過定義接口從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ù)處理框架不能滿足需求，而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è)備通過將下位機(jī)中的采集和運(yùn)行數(shù)據(jù)，通過PLC數(shù)據(jù)采集軟件將采集的數(shù)據(jù)由Netty服務(wù)器調(diào)用RoketMQ生產(chǎn)者接口寫入RoketMQ集群節(jié)點(diǎn)[8]。RoketMQ集群節(jié)點(diǎn)對(duì)持續(xù)實(shí)時(shí)上報(bào)的數(shù)據(jù)，根據(jù)不同告警類型的PLC數(shù)據(jù)寫入相應(yīng)的Topic中，經(jīng)過相關(guān)業(yè)務(wù)處理后的數(shù)據(jù)去向分為3類，第1類是實(shí)時(shí)告警信息以主動(dòng)的方式可通過消息系統(tǒng)SMS(如：移動(dòng)、聯(lián)通、電信、網(wǎng)通、阿里云、微信、騰訊等)，將告警信息及時(shí)發(fā)送到相關(guān)負(fù)責(zé)人，使得告警能夠及時(shí)快速響應(yīng)；第2類是將實(shí)時(shí)上報(bào)的PLC數(shù)據(jù)存入Elasticsearch數(shù)據(jù)庫中，由于Elasticsearch具有很強(qiáng)的數(shù)據(jù)檢索能力，系統(tǒng)在數(shù)據(jù)查詢時(shí)時(shí)效性很高；第3類是將復(fù)雜業(yè)務(wù)處理的數(shù)據(jù)傳遞到Spark Streaming進(jìn)行集群運(yùn)算，并將處理結(jié)果持久化到時(shí)序數(shù)據(jù)庫和關(guān)系數(shù)據(jù)庫中，該類數(shù)據(jù)庫可作為水利自動(dòng)化上報(bào)的成果數(shù)據(jù)庫，為系統(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)定等問題，此時(shí)需要對(duì)集群進(jìn)行調(diào)優(yōu)才能達(dá)到最佳性能。調(diào)優(yōu)是一個(gè)具體性很強(qiáng)的事情，不同的應(yīng)用及場景優(yōu)化方式也不同，調(diào)優(yōu)并沒有統(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ù)過小導(dǎo)致每個(gè)Partition的數(shù)據(jù)量太大，導(dǎo)致內(nèi)存壓力過大，Executor的計(jì)算能力不能充分利用；若Partiton個(gè)數(shù)過多，導(dǎo)致分片過多，執(zhí)行效率低。

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

數(shù)據(jù)序列化在分布式數(shù)據(jù)處理過程中對(duì)集群性能的影響較為顯著。本文主要是采用Java的對(duì)象輸入流框架Object Output Stream Framework進(jìn)行序列化，通過采用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ù)的頻率過高于集群處理速度則將產(chǎn)生數(shù)據(jù)積壓，上報(bào)的PLC數(shù)據(jù)過低于集群處理速度造成算力浪費(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ù)據(jù)研究重要的領(lǐng)域。根據(jù)國家水利部2019年發(fā)布的智慧水利建設(shè)指導(dǎo)意見，要求構(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)測試證明構(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)存的占用率較高的問題，通過對(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)化以滿足實(shí)際項(xiàng)目建設(shè)需要。