基于Spark Streaming的實時能耗分項計量系統(tǒng)

武志學(xué)

1.成都五舟漢云科技有限公司,成都 611731; 2.成都信息工程大學(xué) 信息安全工程學(xué)院,成都 610225)(*通信作者電子郵箱zhixue.wu@gmail.com)

基于Spark Streaming的實時能耗分項計量系統(tǒng)

武志學(xué)1,2*

1.成都五舟漢云科技有限公司,成都 611731; 2.成都信息工程大學(xué) 信息安全工程學(xué)院,成都 610225)(*通信作者電子郵箱zhixue.wu@gmail.com)

能耗分項計量能夠準(zhǔn)確、及時、有效地發(fā)現(xiàn)能源使用問題,形成和實現(xiàn)最有效的節(jié)能措施。能耗分項計量系統(tǒng)需要對各項能源使用量在不同粒度上進(jìn)行統(tǒng)計,既有實時性的需求,又需要涉及到聚合、去重、連接等較為復(fù)雜的統(tǒng)計需求。由于數(shù)據(jù)產(chǎn)生快、實時性強(qiáng)、數(shù)據(jù)量大,所以很難統(tǒng)一采集并入庫存儲后再作處理,這便導(dǎo)致傳統(tǒng)的數(shù)據(jù)處理架構(gòu)不能滿足需求。為此,提出基于Spark Streaming大數(shù)據(jù)流式技術(shù)構(gòu)建一個實時能耗分項計量系統(tǒng),對實時能耗分項計量的系統(tǒng)架構(gòu)和內(nèi)部結(jié)構(gòu)進(jìn)行了詳細(xì)介紹,并通過實驗數(shù)據(jù)分析了系統(tǒng)的實時數(shù)據(jù)處理能力。與傳統(tǒng)架構(gòu)不同,實時能耗分項計量系統(tǒng)在數(shù)據(jù)流動的過程中實時地進(jìn)行捕捉和處理,一方面把捕捉到的異常信息及時報警到前端,同時把分類分項統(tǒng)計處理的結(jié)果保存到數(shù)據(jù)庫,以便進(jìn)行離線分析和數(shù)據(jù)挖掘，能有效地解決上述數(shù)據(jù)處理過程中遇到的問題。

流式計算;能耗分項計量;Spark Streaming;Apache Kafka;大數(shù)據(jù)

0 引言

伴隨著我國城市化進(jìn)程的加快,大型公共建筑節(jié)能工作勢在必行。如何達(dá)到既滿足使用及舒適度的需求,又能科學(xué)、合理地節(jié)能降耗已經(jīng)是全社會所要思考的問題。在大力推廣節(jié)能減排的階段,要達(dá)到最快、最明顯的節(jié)能效果,不單是采用設(shè)備節(jié)能手段,更需要使用分項計量準(zhǔn)確、及時、有效地發(fā)現(xiàn)能源使用問題,形成和實現(xiàn)最有效的節(jié)能措施。能耗分項計量是指對建筑的水、電、燃?xì)狻⒓泄?、集中供冷等各種能耗進(jìn)行監(jiān)測,從而得出建筑物的總能耗量和不同能源種類、不同功能系統(tǒng)的能耗量[1]。要實現(xiàn)分項計量,必須進(jìn)行數(shù)據(jù)采集、數(shù)據(jù)傳輸、數(shù)據(jù)存儲和數(shù)據(jù)分析等。所以,能耗分項計量是一個典型的流式大數(shù)據(jù)系統(tǒng),具有數(shù)據(jù)量大、數(shù)據(jù)產(chǎn)生速度快、數(shù)據(jù)結(jié)構(gòu)復(fù)雜等特點。

一般情況下,能耗分項計量包括空調(diào)系統(tǒng)、電梯系統(tǒng)、給排水系統(tǒng)、通風(fēng)系統(tǒng)、照明系統(tǒng)及辦公設(shè)備系統(tǒng)等。對于用能密度高、單體設(shè)備耗能大的集中空調(diào)系統(tǒng),應(yīng)進(jìn)行更細(xì)致的計量,包括:冷凍主機(jī)用電量、冷凍水泵用電量、冷卻水泵用電量、冷卻塔風(fēng)機(jī)用電量、空調(diào)箱和新風(fēng)機(jī)用電量等。所以進(jìn)行能耗分項計量時,需要對各項能源使用量在不同粒度上對不同數(shù)據(jù)進(jìn)行統(tǒng)計,既有實時性的需求,又需要涉及到聚合、去

圖1 基于傳統(tǒng)數(shù)據(jù)處理模式的分項計量系統(tǒng)

重、連接等較為復(fù)雜的統(tǒng)計需求。由于數(shù)據(jù)產(chǎn)生快、實時性強(qiáng)、數(shù)據(jù)量大,如果采取傳統(tǒng)的數(shù)據(jù)處理架構(gòu),首先對采集到的數(shù)據(jù)入庫存儲,然后再作處理,很難滿足分項計量的需求。特別是為了找到能耗使用規(guī)律提出有效節(jié)能措施,不但需要部署大量能耗采集儀表,還需要進(jìn)行更為復(fù)雜的數(shù)據(jù)處理,從而引起在單位時間內(nèi)要處理的實時數(shù)據(jù)量和計算工作量同時大幅上升，這便導(dǎo)致傳統(tǒng)的數(shù)據(jù)處理架構(gòu)不能滿足需要。為了解決這個問題,本文通過使用Apache Kafka和Spark Streaming模塊構(gòu)建了一個實時流式數(shù)據(jù)處理系統(tǒng)來進(jìn)行能耗分項計量。與傳統(tǒng)架構(gòu)不同,實時流式數(shù)據(jù)處理系統(tǒng)在數(shù)據(jù)流動的過程中實時地進(jìn)行捕捉和處理,并根據(jù)業(yè)務(wù)需求對數(shù)據(jù)進(jìn)行計算分析,一方面把捕捉到的異常信息及時報警到前端,同時把分類分項統(tǒng)計處理的結(jié)果保存到數(shù)據(jù)庫,以便進(jìn)行離線分析和數(shù)據(jù)挖掘。本文將詳細(xì)描述實時能耗分項計量的系統(tǒng)架構(gòu)和內(nèi)部結(jié)構(gòu),并對架構(gòu)中所使用的大數(shù)據(jù)技術(shù)和系統(tǒng)進(jìn)行介紹和分析,最后,通過實際測試結(jié)果對實時能耗分項計量系統(tǒng)的實時數(shù)據(jù)處理能力進(jìn)行驗證和分析。

1 相關(guān)研究

清華大學(xué)節(jié)能研究中心研制開發(fā)了能耗分項計量實時分析系統(tǒng)EM-II[2],包括數(shù)據(jù)采集子系統(tǒng)、數(shù)據(jù)處理子系統(tǒng)、數(shù)據(jù)分析展示子系統(tǒng)三大核心部分,另外還有信息維護(hù)、數(shù)據(jù)上報、系統(tǒng)監(jiān)測等幾個子系統(tǒng)。數(shù)據(jù)采集子系統(tǒng)利用安裝在現(xiàn)場的具有數(shù)字通信接口的電計量表和超聲波冷熱量表采集數(shù)據(jù),并由數(shù)據(jù)采集器匯總接收通過網(wǎng)關(guān)由路由器連接到互聯(lián)網(wǎng),將數(shù)據(jù)遠(yuǎn)程傳輸回數(shù)據(jù)中心服務(wù)器。數(shù)據(jù)處理子系統(tǒng)負(fù)責(zé)校驗解析接收到的原始數(shù)據(jù), 并根據(jù)能耗模型拆分計算得到分類分項數(shù)據(jù)。數(shù)據(jù)分析展示子系統(tǒng)將經(jīng)過數(shù)據(jù)處理后的分類分項能耗數(shù)據(jù)進(jìn)行分析、匯總和整合，一方面通過靜態(tài)或者動態(tài)的圖表方式將能耗數(shù)據(jù)展示出來,另一方面能夠提供針對第三方的數(shù)據(jù)接入服務(wù)和數(shù)據(jù)發(fā)布服務(wù)。

Hysine與多個高等院校及科研機(jī)構(gòu)合作研制開發(fā)的EMC-2000建筑設(shè)備節(jié)能控制與管理系統(tǒng)[3],適用于新建、改建、擴(kuò)建項目中建筑機(jī)電設(shè)備能效跟蹤控制節(jié)能管理。整個能源管理系統(tǒng)由管理中心、主干通信網(wǎng)絡(luò)、數(shù)據(jù)采集器、智能電表等組成,同時為與上一級能耗監(jiān)測和管理系統(tǒng)連接預(yù)留系統(tǒng)接口。能源管理中心通過對現(xiàn)場數(shù)據(jù)采集器上傳的數(shù)據(jù)進(jìn)行存儲、統(tǒng)計和分析,為業(yè)主提供有效的能源使用和持續(xù)的能源節(jié)約提供實施依據(jù)。

安科瑞開發(fā)的Acrel-5000建筑能耗分析管理系統(tǒng)[4]以計算機(jī)、通信設(shè)備、測控單元為基本工具,根據(jù)現(xiàn)場實際情況采用現(xiàn)場總線、光纖環(huán)網(wǎng)或無線通信中的一種或多種結(jié)合的最優(yōu)化的組網(wǎng)方式,為大型公共建筑的實時數(shù)據(jù)采集及遠(yuǎn)程管理與控制提供了基礎(chǔ)平臺,它可以和檢測設(shè)備構(gòu)成任意復(fù)雜的監(jiān)控系統(tǒng)。

這些能耗分項計量系統(tǒng)都是參照國家住建部《國家機(jī)關(guān)辦公建筑和大型公共建筑能耗監(jiān)測系統(tǒng)》[5]和《國家機(jī)關(guān)辦公建筑和大型公共建筑能耗監(jiān)測系統(tǒng)省、市級數(shù)據(jù)中心數(shù)據(jù)庫結(jié)構(gòu)文檔》[6],采用了傳統(tǒng)的數(shù)據(jù)處理模式,如圖1所示。當(dāng)數(shù)據(jù)采集程序接收到數(shù)據(jù)采集器發(fā)送的數(shù)據(jù)以后,首先把數(shù)據(jù)寫入計量表原始數(shù)值數(shù)據(jù)庫(D)；然后再由拆分程序按照各個儀表和能耗數(shù)據(jù)各級分項進(jìn)行拆分和統(tǒng)計,并把結(jié)果寫入分類分項能耗數(shù)據(jù)庫(B)；最后再由分析展示程序基于建筑基本情況數(shù)據(jù)庫(A)、分類分項能耗數(shù)據(jù)庫(B)進(jìn)行數(shù)據(jù)分析并展示給用戶。

這種基于傳統(tǒng)數(shù)據(jù)處理模式的分項計量系統(tǒng)只能適用于采集點數(shù)量比較少、統(tǒng)計分析比較簡單的環(huán)境。在采集點數(shù)量達(dá)到上千時,隨著時間的推移,分類分項能耗數(shù)據(jù)庫的數(shù)據(jù)會不斷累計快速增加,從而可能導(dǎo)致拆分程序無法及時完成對數(shù)據(jù)的拆分和統(tǒng)計。

2 實時能耗分項計量系統(tǒng)

為了解決基于傳統(tǒng)數(shù)據(jù)處理模式的能耗分項計量系統(tǒng)存在的問題,本文設(shè)計并實現(xiàn)了一個基于Spark Streaming和Apache Kafka等大數(shù)據(jù)技術(shù)的實時能耗分項計量系統(tǒng)。在本章首先對Spark Streaming和Apache Kafka大數(shù)據(jù)技術(shù)進(jìn)行簡單介紹,然后描述如何使用Spark Streaming和Apache Kafka模塊構(gòu)建基于實時流式數(shù)據(jù)處理架構(gòu)的實時能耗分項計量系統(tǒng)。

2.1 Spark Streaming

Apache Spark是一個基于內(nèi)存的、可以支持各種大數(shù)據(jù)應(yīng)用場景的、高性能和高容錯的開源大數(shù)據(jù)平臺[7]。Spark Streaming是Apache Spark的一個子項目,是一個運(yùn)行在Spark引擎之上的實時處理工具[8]。

與Hadoop[9]大數(shù)據(jù)處理平臺不同,Spark建立在統(tǒng)一抽象的RDD(Resilient Distributed Datasets)之上,使得它可以以基本一致的方式應(yīng)對各種大數(shù)據(jù)處理場景,包括MapReduce、Streaming、SQL、Machine Learning以及Graph等。

Spark的另一個特點就是其高性能和容錯性。Spark是一種粗粒度數(shù)據(jù)并行的計算范式，計算的主體是數(shù)據(jù)集合RDD,而非個別數(shù)據(jù)。RDD集合內(nèi)的所有數(shù)據(jù)都經(jīng)過同樣的算子序列，數(shù)據(jù)并行可編程性好,易于獲得高并行性(與數(shù)據(jù)規(guī)模相關(guān),而非與程序邏輯的并行性相關(guān)),也易于高效地映射到底層的并行或分布式硬件上[10]。RDD是一個容錯的、并行的數(shù)據(jù)結(jié)構(gòu),在保證容錯的前提下,用內(nèi)存來承載工作集。內(nèi)存的存取速度快于磁盤多個數(shù)量級,從而可以極大提升性能[11]。Spark的容錯是通過重放日志更新而取得的。因為Spark的函數(shù)式語義和冪等特性,重放日志更新RDD不會有副作用。另外,Spark記錄的是粗粒度的RDD更新,所以容錯的開銷可以忽略不計。

Spark的實時性特點是通過Spark Streaming實現(xiàn)的。Spark Streaming將流式計算分解成一系列短小的批處理作業(yè),也就是把輸入數(shù)據(jù)流按照批次大小(如1 s)分成一段一段的數(shù)據(jù)形成DStream(Discretized Stream),而每一段數(shù)據(jù)都轉(zhuǎn)換成Spark中的RDD,如圖2所示。

圖2 DStream的組成

Spark Streaming提供了兩種操作類型,分別為Transformations和Output操作。對DStream的Transformation操作變?yōu)镾park中對RDD的Transformation操作,從一個已知的DStream經(jīng)過轉(zhuǎn)換得到一個新的DStream；而且Spark Streaming還額外增加了一類針對“窗口”(Window)的Transformation操作,可以更靈活地控制DStream的大小(時間間隔大小、數(shù)據(jù)元素個數(shù)等)。整個流式計算根據(jù)業(yè)務(wù)的需求可以對中間的結(jié)果進(jìn)行疊加,或者使用Output操作將DStream數(shù)據(jù)輸出到一個外部的存儲系統(tǒng),如數(shù)據(jù)庫或文件系統(tǒng)等。

Spark具有極高的擴(kuò)展性與吞吐量。根據(jù)Spark官方網(wǎng)站FAQ,最大的已知Spark集群有8 000個節(jié)點[12]；并且隨著大數(shù)據(jù)增多,預(yù)計集群規(guī)模也會隨之變大,以便繼續(xù)滿足吞吐量方面的預(yù)期。另外,使用Spark的EC2 啟動腳本可以輕松地在Amazon EC2上啟動一個獨立集群。Spark目前在EC2上已能夠線性擴(kuò)展到100個節(jié)點(每個節(jié)點4核),可以以數(shù)秒的延遲每秒處理6 GB的數(shù)據(jù)量[11]。

一個Spark集群由多個工作節(jié)點(Worker Node)組成,每個工作節(jié)點可以運(yùn)行一個或多個Executor,如圖3所示。Executor是一個用于應(yīng)用程序或者工作節(jié)點的進(jìn)程,負(fù)責(zé)處理Tasks,并將數(shù)據(jù)保存到內(nèi)存或者磁盤中。每個應(yīng)用程序都有屬于自己的Executor,一個Executor則包含了一定數(shù)量的Slots來運(yùn)行分配給它的任務(wù)。在Spark中,每個作業(yè)(Job)都被分隔成多個彼此依賴稱之為Stage的Task。 一個Task就是一個工作單元, 可以發(fā)送給一個Executor執(zhí)行。 Task是用來執(zhí)行應(yīng)用的實際計算工作。 每個Task占用Executor的一個Slot。

圖3 Spark Streaming集群架構(gòu)

Spark Streaming流計算可以在數(shù)據(jù)產(chǎn)生并流入系統(tǒng)時就進(jìn)行處理并馬上得出結(jié)果,非常適合能耗分項計量中能耗數(shù)據(jù)不斷產(chǎn)生的場景和對實時性的需求。

選擇Spark Streaming的另一個原因是因為Spark可以在支持實時流式處理的同時,進(jìn)行高效的批處理、交互式SQL查詢和數(shù)據(jù)挖掘,從而可以使能耗分項計量系統(tǒng)不但可以實時地為用戶捕捉能耗異常情況進(jìn)行報警,還可以提供離線統(tǒng)計分析和數(shù)據(jù)挖掘的服務(wù)。

2.2 Apache Kafka

Apache Kafka是一個分布式的、高吞吐量的、易于擴(kuò)展的基于主題發(fā)布/訂閱的消息系統(tǒng),最早是由 LinkedIn 開發(fā),并于2011年開源并貢獻(xiàn)給 Apache 軟件基金會[13]。

Kafka的邏輯架構(gòu)如圖4所示。Kafka對消息保存時根據(jù)話題(Topic)進(jìn)行歸類,發(fā)送消息者成為生產(chǎn)者(Producer),消息接受者成為消費者(Consumer)。此外Kafka集群由多個服務(wù)器組成,每個服務(wù)器成為代理(Broker)。無論是Kafka集群,還是Producer和Consumer都依賴于Zookeeper來保證系統(tǒng)可用性。

圖4 Kafka的邏輯架構(gòu)

一個話題可以認(rèn)為是一類消息,每個話題將被分成多個分區(qū)(Partition)。設(shè)計分區(qū)的最根本原因是Kafka基于文件存儲,通過分區(qū)可以將日志內(nèi)容分散到多個服務(wù)器上,來均衡負(fù)載,保證消息保存/消費的效率。如果一個話題對應(yīng)一個文件,那這個文件所在的機(jī)器IO將會成為這個話題的性能瓶頸,而有了分區(qū)后,不同的消息可以并行寫入不同代理的不同分區(qū)里,屬于順序?qū)懘疟P,因此效率非常高,極大地提高了Kafka的吞吐率。所以,消息分區(qū)是Kafka高吞吐率的一個很重要的保證，即使在非常廉價的商用機(jī)器上也能做到單機(jī)支持每秒104條以上消息的傳輸[14]。此外,越多的分區(qū)意味著可以容納更多的消費者,可以有效提升并發(fā)消費的能力。Kafka的消息分區(qū)結(jié)構(gòu)如圖5所示。

圖5 Kafka消息分區(qū)結(jié)構(gòu)

與傳統(tǒng)的消息系統(tǒng)不同,Kafka系統(tǒng)中存儲的消息沒有明確的消息ID，消息通過日志中的位置稱為偏移量來唯一標(biāo)記一條消息，這樣就避免了維護(hù)密集尋址,用于映射消息ID到實際消息地址的隨機(jī)存取索引結(jié)構(gòu)的開銷。這種設(shè)計大大提高了Kafka的性能。

Kafka的另外一個創(chuàng)新是即使消息被消費,消息仍然不會被立即刪除。日志文件將會根據(jù)代理中的配置,保留一定的時間之后刪除;比如日志文件保留2 d,那么之后文件會被清除,無論其中的消息是否被消費。Kafka通過這種簡單的手段來釋放磁盤空間,從而可以減少消息消費之后對文件內(nèi)容改動的磁盤IO開支。

Kafka還有一個創(chuàng)新點就是Kafka代理是無狀態(tài)的,由消費者維護(hù)已消費的狀態(tài)信息。這種設(shè)計的一個好處就是消費者可以退回到老的偏移量再次消費數(shù)據(jù)。因為代理是無狀態(tài)的,它不需要標(biāo)記哪些消息被哪些消費者消費過,也不需要代理去保證同一個消費者組里只有一個消費者能消費某一條消息,因此也就不需要鎖機(jī)制,這也為Kafka的高吞吐率提供了有力保障。

為了提高可用性,Kafka可以配置分區(qū)需要備份的個數(shù),每個分區(qū)將會被備份到多臺Kafka服務(wù)器上,以提高可用性。每個分區(qū)都有一個Kafka服務(wù)器為領(lǐng)導(dǎo)者(Leader),負(fù)責(zé)所有的讀寫操作。如果領(lǐng)導(dǎo)者失效,那么將會有其他跟隨者(Follower)來接管成為新的領(lǐng)導(dǎo)者。跟隨者只是單調(diào)地和領(lǐng)導(dǎo)者跟進(jìn),同步消息即可。從集群的整體考慮,Kafka會將領(lǐng)導(dǎo)者均衡地分散到每個Kafka服務(wù)器上,來確保整體的性能穩(wěn)定。

Kafka可以同時支持離線數(shù)據(jù)處理和實時數(shù)據(jù)處理。如圖6所示,Kafka同時支持點到點分發(fā)模型,即多個消費者共同消費隊列中某個消息的單個副本,以及發(fā)布-訂閱模型,即多個消費者接收自己的消息副本。根據(jù)這一特性,可以使用Spark實時流處理系統(tǒng)對消息進(jìn)行實時在線處理,同時使用Hadoop這種批處理系統(tǒng)進(jìn)行離線處理,還可以同時將數(shù)據(jù)進(jìn)行實時備份,只需要保證這三個操作所使用的消費者屬于不同的消費者即可。

圖6 Kafka消息分發(fā)模型

總之,Kafka是一種處理大量數(shù)據(jù)的新型消息系統(tǒng),其高吞吐量、高可靠、高可用、易擴(kuò)展的特性完全適應(yīng)于能耗分項計量系統(tǒng)。此外,通過利用Kafka同時支持多種處理模型的特點,能耗分項計量系統(tǒng)可以在進(jìn)行能耗數(shù)據(jù)在線處理的同時,對能耗數(shù)據(jù)進(jìn)行備份和離線處理。

2.3 實時能耗分項計量系統(tǒng)架構(gòu)

本文設(shè)計的實時能耗分項計量系統(tǒng)的整體架構(gòu)如圖7所示,主要包括后端能耗數(shù)據(jù)采集部分、Kafka消息系統(tǒng)、Spark Streaming集群、Hadoop集群、前端實時展示應(yīng)用和前端分析展示應(yīng)用,以及分類分項能耗數(shù)據(jù)庫和計量表原始數(shù)值數(shù)據(jù)庫。

流式處理系統(tǒng)主要通過網(wǎng)絡(luò)Socket通信來實現(xiàn)與外部IO系統(tǒng)的數(shù)據(jù)交互。由于網(wǎng)絡(luò)通信的不可靠特點,發(fā)送端與接收端需要通過一定的協(xié)議來保證數(shù)據(jù)包的接收確認(rèn)和失敗重發(fā)機(jī)制。不是所有的IO系統(tǒng)都支持重發(fā),這至少需要實現(xiàn)數(shù)據(jù)流的持久化,同時還要實現(xiàn)高吞吐和低時延。通過前面的介紹,可以確定Kafka具備這些特點,完全能夠作為實時能耗分項計量系統(tǒng)的外部數(shù)據(jù)源。

除了把Kafka當(dāng)成輸入數(shù)據(jù)源之外,實時能耗分項計量系統(tǒng)還將其作為信息輸出數(shù)據(jù)源向前端實時展示應(yīng)用推送相關(guān)報警和實時流信息。

圖7 實時能耗分項計量系統(tǒng)整體架構(gòu)

能耗數(shù)據(jù)采集部分包括能耗采集儀表和數(shù)據(jù)采集器。一般來講,每個數(shù)據(jù)采集器可以連接16到32塊采集儀表。數(shù)據(jù)采集器負(fù)責(zé)接收所連接采集儀表發(fā)來的能耗數(shù)據(jù),并把數(shù)據(jù)整理為住建部所制定的能耗數(shù)據(jù)通信協(xié)議格式[15],然后按照設(shè)置的時間間隔發(fā)送到設(shè)定的數(shù)據(jù)接收器。每個數(shù)據(jù)接收器就是Kafka消息系統(tǒng)的消息生產(chǎn)者,負(fù)責(zé)把從數(shù)據(jù)采集器發(fā)來的數(shù)據(jù)寫入Kafka消息系統(tǒng),從而保證了數(shù)據(jù)的可靠性。

按照住建部要求,計量表采集到的能耗數(shù)據(jù)一方面必須寫入計量表原始數(shù)值數(shù)據(jù)庫,同時還需要按儀表、按分項進(jìn)行拆分并把結(jié)果寫入分類分項能耗數(shù)據(jù)庫。在滿足住建部基本要求的同時,實時能耗分項計量系統(tǒng)還對能耗數(shù)據(jù)進(jìn)行實時分析以便能夠及時捕捉能耗異常情況,并報警給用戶。

為了能夠保證能耗數(shù)據(jù)處理的實時性,實時能耗分項計量系統(tǒng)充分利用Kafka消息系統(tǒng)可以同時支持多個消費者組的能力,為能耗數(shù)據(jù)消息設(shè)置兩個消費者組。一個是運(yùn)行在Spark Streaming流式計算集群上的能耗數(shù)據(jù)實時數(shù)據(jù)拆分程序;另一個則是運(yùn)行在Spark批處理集群上的計量表原始數(shù)值寫入程序。

運(yùn)行在Spark Streaming集群上的能耗數(shù)據(jù)拆分程序是實時能耗分項計量系統(tǒng)的核心模塊。數(shù)據(jù)拆分程序以Kafka消息系統(tǒng)作為能耗數(shù)據(jù)輸入流進(jìn)行實時在線處理。首先,數(shù)據(jù)拆分程序?qū)δ芎臄?shù)據(jù)進(jìn)行分類分項拆分,并形成多個數(shù)據(jù)流供其他業(yè)務(wù)處理模塊使用。第二,數(shù)據(jù)拆分程序把分類分項拆分結(jié)果按照不同時間粒度進(jìn)行統(tǒng)計,并把統(tǒng)計結(jié)果寫入分類分項能耗數(shù)據(jù)庫。時間粒度分為15 min、小時、天和月。

除了能耗數(shù)據(jù)拆分程序之外,Spark Streaming集群還可以進(jìn)行多種實時在線數(shù)據(jù)處理,比如能耗熱點分析和能耗異常分析。這些能耗數(shù)據(jù)處理程序并不直接從Kafka消息系統(tǒng)中獲取數(shù)據(jù),而是使用能耗數(shù)據(jù)拆分程序生成的數(shù)據(jù)流進(jìn)行數(shù)據(jù)處理,并把分析結(jié)果通過Kafka消息系統(tǒng)提供給前端實時展示應(yīng)用。

實時能耗分項計量系統(tǒng)的另一部分功能是進(jìn)行離線數(shù)據(jù)統(tǒng)計、分析以及數(shù)據(jù)挖掘。前端能耗分析程序基于能耗數(shù)據(jù)拆分程序?qū)懭氲椒诸惙猪椖芎臄?shù)據(jù)庫的數(shù)據(jù),以及計量表原始數(shù)值數(shù)據(jù)進(jìn)行各類能耗數(shù)據(jù)統(tǒng)計、分析以及數(shù)據(jù)挖掘,從而使用能單位可以掌握詳細(xì)能耗使用情況,為制定節(jié)能策略提供科學(xué)依據(jù)。

因為能耗數(shù)據(jù)記錄的數(shù)量遠(yuǎn)遠(yuǎn)超過了傳統(tǒng)關(guān)系型數(shù)據(jù)庫可以支持的容量,計量表原始數(shù)值數(shù)據(jù)庫和分類分項能耗數(shù)據(jù)庫均使用了HBase數(shù)據(jù)庫。HBase是運(yùn)行在Hadoop上的NoSQL數(shù)據(jù)庫,它是一個分布式的和可擴(kuò)展的面向列的數(shù)據(jù)庫,可以在一組通用硬件上存儲許多具有數(shù)十億行和上百萬列的大表[16]。HBase能夠融合Key-Value數(shù)據(jù)模式帶來實時查詢的能力,以及通過MapReduce或Spark進(jìn)行離線處理或者批處理的能力。總之,HBase能夠存儲大量的數(shù)據(jù),讓用戶在大量的數(shù)據(jù)中查詢記錄,并從中獲得綜合分析報告。所以,HBase非常適合于存放計量表原始數(shù)據(jù)和分類分項能耗數(shù)據(jù)。HBase不但可以滿足能耗分項計量系統(tǒng)每天幾十萬條記錄的大數(shù)據(jù)量需求,還可以與Hadoop的MapReduce以及Spark SQL和Spark MLlib結(jié)合為用戶提供高效能耗數(shù)據(jù)分析和數(shù)據(jù)挖掘工作。

2.4 實時能耗分項計量系統(tǒng)內(nèi)部結(jié)構(gòu)

實時能耗分項計量系統(tǒng)的內(nèi)部結(jié)構(gòu)如圖8所示。Spark Streaming集群由多個工作者節(jié)點(Worker Node)組成,每個工作者節(jié)點包含一個或多個Spark Executor。同時,在每個工作節(jié)點還安裝了用來存儲能耗數(shù)據(jù)的數(shù)據(jù)庫系統(tǒng)HBase和MySQL,以及數(shù)據(jù)倉庫系統(tǒng)Hive。HBase用來存儲能耗分類分項數(shù)據(jù)和計量表原始數(shù)值數(shù)據(jù);MySQL用來存儲與用能單位和分項計量系統(tǒng)各種設(shè)備部署情況的結(jié)構(gòu)化數(shù)據(jù)；Hive用來按主題、多維度、多粒度對分類分項能耗數(shù)據(jù)進(jìn)行存儲和管理,為后期進(jìn)行離線分析和數(shù)據(jù)挖掘提供良好的基礎(chǔ)。

圖8 實時能耗分項計量系統(tǒng)內(nèi)部結(jié)構(gòu)

來自各個數(shù)據(jù)采集器的能耗數(shù)據(jù)使用同一個話題(Topic)寫入Kafka消息系統(tǒng)。為了保證實時能耗分項計量系統(tǒng)的吞吐量,以便支持更多的計量表,我們對能耗數(shù)據(jù)消息話題進(jìn)行了分區(qū)。通過消息分區(qū)可以提高消息生產(chǎn)者和消息消費者的并發(fā)能力。

在實時能耗分項計量系統(tǒng)中,每個消息分區(qū)將有兩個消息消費者小組:一個是Spark Streaming流式集群的能耗數(shù)據(jù)拆分程序小組;另一個是Spark批處理集群的計量表原始數(shù)值數(shù)據(jù)寫入程序小組。每個小組都由多個消費者組成,每個消息分區(qū)的數(shù)據(jù)都會被每個小組中的一個消費者接收。

圖8描述了一個包含兩個Worker Node的Spark 集群,每個Worker Node運(yùn)行了三個Spark Executor。每個Worker Node上有兩個Executor屬于能耗數(shù)據(jù)拆分程序小組,另外一個屬于計量表原始數(shù)值寫入程序小組。

能耗數(shù)據(jù)消息話題分成了四個分區(qū),能耗數(shù)據(jù)拆分程序小組中的每個Executor消費一個消息分區(qū)的數(shù)據(jù);計量表原始數(shù)據(jù)值寫入程序小組中的每個Executor負(fù)責(zé)消費兩個消息分區(qū)的數(shù)據(jù)。一般來講,分區(qū)的個數(shù)最好是消費者小組中消費者的倍數(shù),也就是說,同小組中的每個消費者負(fù)責(zé)處理的消息分區(qū)個數(shù)是等同的。在實際環(huán)境中消息話題分區(qū)的個數(shù)需要按照整個系統(tǒng)連接的計量儀表的個數(shù)來確定。

與Receiver方式相比，Direct方式雖然使用較為復(fù)雜，但是它能提供更好的靈活性和可靠性，所以本文選用Direct方式。Direct方式使用Kafka的基本API,由Spark Streaming負(fù)責(zé)記錄在每個消息分區(qū)中的消費位移,也就是已經(jīng)消費過的消息位置,并保存在Spark系統(tǒng)的檢測點(Check Point)記錄中。使用Direct方式,Spark Streaming會周期性地查詢Kafka,來獲得每個消息分區(qū)的最新的位移,從而定義每個數(shù)據(jù)塊的數(shù)據(jù)范圍。當(dāng)處理消息的作業(yè)啟動時,就會使用Kafka的簡單消費API 來獲取Kafka指定范圍的數(shù)據(jù)。Spark會創(chuàng)建跟Kafka 分區(qū)一樣多的RDD 分區(qū),并且會并行從Kafka中讀取數(shù)據(jù)。所以在Kafka 分區(qū)和RDD 分區(qū)之間,有一個一對一的映射關(guān)系。采用Direct方式的另外一個優(yōu)勢就是可以利用Kafka保證數(shù)據(jù)的可靠性,并且可以保證數(shù)據(jù)是消費一次且僅消費一次。

在每個Spark Executor中,運(yùn)行著能耗數(shù)據(jù)拆分程序和多個能耗數(shù)據(jù)實時處理程序。能耗拆分程序的功能在前面已經(jīng)介紹。每個能耗數(shù)據(jù)處理程序基于拆分程序生成的實時數(shù)據(jù)流完成一定的數(shù)據(jù)處理工作,并把部分?jǐn)?shù)據(jù)處理的結(jié)果寫入Kafka消息系統(tǒng)供前端實時展示應(yīng)用使用,同時還會把一些數(shù)據(jù)處理結(jié)果寫入MySQL數(shù)據(jù)庫供前端分析系統(tǒng)使用。

能耗用量異常分析程序是我們提供的一個能耗實時數(shù)據(jù)處理程序,它基于數(shù)據(jù)拆分程序提供的能耗使用數(shù)據(jù)流,根據(jù)用能單位設(shè)置的各類閾值以及正常能耗使用量發(fā)現(xiàn)用能異常情況,并通過Kafka消息系統(tǒng)及時報警給前端實時展示應(yīng)用。比如,單位給某辦公樓層設(shè)置的空調(diào)用電的閾值為每小時20度,能耗異常分析程序在對能耗使用數(shù)據(jù)流進(jìn)行處理時就會檢測該樓層的空調(diào)用電量,當(dāng)用電量超過每小時20度時,就會產(chǎn)生報警消息通過Kafka提交給前端實時展示應(yīng)用。同樣的,假定某小區(qū)正常煤氣流量為10 m3/min左右,如果能耗用量異常分析程序發(fā)現(xiàn)該小區(qū)煤氣流量遠(yuǎn)遠(yuǎn)超過了10 m3/min,那么就有可能是發(fā)生了煤氣管道漏氣。這時,能耗用量異常分析程序就會產(chǎn)生報警消息。

能耗用量熱點分析程序是我們提供的另一個能耗實時數(shù)據(jù)處理程序,它會實時統(tǒng)計每個計量點的每刻的能耗使用量并通過Kafka消息系統(tǒng)發(fā)布。前端實時展示應(yīng)用可以獲取感興趣的計量點的流量統(tǒng)計來繪制能耗用量熱點圖,從而可以一目了然地及時了解所關(guān)心計量點的能耗使用狀況。

2.5 實時能耗分項計量系統(tǒng)的優(yōu)勢

本文提出的實時能耗分項計量系統(tǒng)充分利用了最先進(jìn)的大數(shù)據(jù)技術(shù),特別是流計算技術(shù),并針對能耗分項計量的特點對整體系統(tǒng)架構(gòu)和內(nèi)部結(jié)構(gòu)進(jìn)行了認(rèn)真的研究與設(shè)計。與傳統(tǒng)的能耗分項計量系統(tǒng)比,本文提出的實時能耗分項計量系統(tǒng)具有如下優(yōu)勢：

首先,實時能耗分項計量系統(tǒng)可以同時支持實時在線數(shù)據(jù)處理和離線數(shù)據(jù)統(tǒng)計分析,而傳統(tǒng)的能耗分項計量系統(tǒng)只支持對能耗使用情況的離線統(tǒng)計和分析。實時能耗分項計量系統(tǒng)的異常情況實時報警功能和能耗使用熱點實時分析功能,不僅可以使用能單位在發(fā)生能耗異常情況時可以及時采取相應(yīng)措施,防止異常情況蔓延,還可以讓用能單位隨時掌握整體能耗情況的實時現(xiàn)狀。

其次,實時能耗分項計量系統(tǒng)具有很強(qiáng)的數(shù)據(jù)處理能力。整體系統(tǒng)架構(gòu)使用了當(dāng)前最先進(jìn)的快速流式處理系統(tǒng)Spark Streaming和具有高可靠、高吞吐量的Kafka消息系統(tǒng)作為實時數(shù)據(jù)流處理的核心架構(gòu)。整個數(shù)據(jù)處理過程是基于內(nèi)存,而不像傳統(tǒng)能耗分項計量系統(tǒng)需要把數(shù)據(jù)首先寫入文件系統(tǒng),然后再讀入到內(nèi)存進(jìn)行處理,所以,實時能耗分項計量系統(tǒng)的處理效率會比傳統(tǒng)能耗分項計量系統(tǒng)提高百倍以上。這意味著,在同樣的硬件配置情況下,實時能耗分析計量系統(tǒng)可以支持的能耗采集點數(shù)可以提高上百倍。

第三,實時能耗分項計量系統(tǒng)具有很強(qiáng)的可擴(kuò)展性。實時能耗分項計量系統(tǒng)架構(gòu)中的Kafka消息系統(tǒng)、Spark系統(tǒng)、HBase系統(tǒng)和Hadoop系統(tǒng)都是分布式集群結(jié)構(gòu),并具有很強(qiáng)的擴(kuò)展能力。所以,在使用實時能耗分項計量系統(tǒng)的每個階段,用戶只需要部署能夠滿足當(dāng)時能耗監(jiān)控需求的設(shè)備即可,而不需要考慮后期可能的需求。這一方面可以節(jié)省用戶的投資成本,還減少了用戶初期部署的設(shè)計負(fù)擔(dān)。

第四,實時能耗分項計量系統(tǒng)提供快速數(shù)據(jù)挖掘能力。除了強(qiáng)大的實時數(shù)據(jù)處理能力以外,借助于Spark 平臺,實時能耗分項計量系統(tǒng)還可以利用Spark MLlib進(jìn)行深度數(shù)據(jù)挖掘,發(fā)現(xiàn)復(fù)雜的能耗數(shù)據(jù)之間的關(guān)聯(lián)關(guān)系,從而為制定有效的節(jié)能措施提供科學(xué)依據(jù)?；赟park MLlib的數(shù)據(jù)挖掘效率會遠(yuǎn)遠(yuǎn)高于基于MapReduce模式的Mahout數(shù)據(jù)挖掘系統(tǒng)的效率。

第五,實時能耗分項計量系統(tǒng)可以很容易增加新的業(yè)務(wù)處理功能。在當(dāng)前系統(tǒng)中,提供了能耗異常分析和能耗用量熱點分析兩個實時處理功能,但是今后可以根據(jù)用戶需求很方便地添加新的業(yè)務(wù)處理能力。新添加的業(yè)務(wù)處理功能將會與原有的處理并行進(jìn)行,并不會影響現(xiàn)有的實時業(yè)務(wù)處理能力。

3 實驗與分析

為了檢驗實時能耗分項計量系統(tǒng)進(jìn)行分項計量和實時數(shù)據(jù)處理的能力,實際部署了一套實時能耗分項計量系統(tǒng),在對各種參數(shù)進(jìn)行優(yōu)化之后,進(jìn)行了一系列的測試。

3.1 測試環(huán)境

測試環(huán)境是運(yùn)行在云平臺上的7臺虛擬機(jī)組成。每臺虛擬機(jī)的配置為8核CPU,25 GB內(nèi)存,1 TB HDD磁盤。圖9描述了實時能耗分項計量系統(tǒng)各個模塊的部署情況。

圖9 系統(tǒng)測試環(huán)境

系統(tǒng)部署的指導(dǎo)思想是要保證整個系統(tǒng)的可靠性和可擴(kuò)充性,并且保證節(jié)點之間的負(fù)載均衡性。具體部署情況如下:在兩臺服務(wù)器上部署了Hadoop的Name Node和HBase的Master Server;三臺服務(wù)器上部署了Zookeeper、Spark、Hadoop的Data Node和HBase的Master Server;最后兩臺服務(wù)器上部署了Kafka、Spark、Hadoop的Data Node和HBase的Master Server。

3.2 測試結(jié)果與分析

實時能耗分項計量系統(tǒng)的性能指標(biāo)主要考慮的是系統(tǒng)的吞吐量和處理數(shù)據(jù)的延遲時間。系統(tǒng)的吞吐量一般用兩個指標(biāo)來衡量：一是單位時間內(nèi)系統(tǒng)能夠處理的能耗數(shù)據(jù)的條數(shù);二是系統(tǒng)處理一條能耗數(shù)據(jù)所需要的時間。單位時間內(nèi)處理的數(shù)據(jù)條數(shù)越多說明系統(tǒng)的吞吐量越高,系統(tǒng)處理數(shù)據(jù)的能力越強(qiáng)。處理能耗數(shù)據(jù)的延遲時間的指標(biāo)也有兩個：一個是從接收到一條能耗數(shù)據(jù)到開始處理該條數(shù)據(jù)之間的時間間隔稱為調(diào)度延遲時間(Scheduling Delay);另一個是從接收到一條能耗數(shù)據(jù)到處理完該條數(shù)據(jù)之間的時間間隔稱為總延遲時間(Total Delay)。處理能耗數(shù)據(jù)的延遲時間越小,說明系統(tǒng)處理數(shù)據(jù)越及時,系統(tǒng)實時性越強(qiáng)。

在測試中,通過給實時能耗分項計量系統(tǒng)的Kafka消息系統(tǒng)加載實際數(shù)據(jù)來測試系統(tǒng)的吞吐量和處理數(shù)據(jù)的延遲時間,測試結(jié)果如圖10所示。圖中展示的測試運(yùn)行了6 min 27 s,每秒加載一組能耗數(shù)據(jù),每組數(shù)據(jù)大約包含140條能耗記錄,總共處理了387組數(shù)據(jù),64 968條能耗記錄。

圖10(a)展示的是給系統(tǒng)加載能耗數(shù)據(jù)的速率(Input Rate)?？梢钥闯鼋o系統(tǒng)加載能耗數(shù)據(jù)的平均速率為每秒167.88條記錄,瞬間最高值達(dá)到了每秒300條以上,絕大多數(shù)數(shù)據(jù)都是按每秒140到200條數(shù)據(jù)的速率發(fā)送的。

圖10(b)展示的是數(shù)據(jù)的調(diào)度延遲時間。盡管顯示的平均調(diào)度延遲時間為10 ms,但從圖中可以看出這主要是由于在測試剛開始啟動時,第一批數(shù)據(jù)有一個2 s延遲而導(dǎo)致的。從右圖可以看出,其余批次數(shù)據(jù)的調(diào)度延遲平均值在0.2 ms以內(nèi)。

圖10(c)展示的是處理一批能耗數(shù)據(jù)所需要的時間。圖中顯示處理每批數(shù)據(jù)的平均時間為133 ms。如果考慮到除去系統(tǒng)剛啟動運(yùn)行的第一批數(shù)據(jù),那么平均處理每一批數(shù)據(jù)的時間會在100 ms以內(nèi)。從圖的形狀來看,除了第一批數(shù)據(jù)以外,系統(tǒng)整個處理過程非常平穩(wěn)。

圖10(d)展示的是系統(tǒng)處理能耗數(shù)據(jù)的總延遲的平均時間為143 ms。類似于調(diào)度延遲時間,總延遲平均時間也因為第一批數(shù)據(jù)的延遲而拉高。如果剔除第一批數(shù)據(jù),其余批次數(shù)據(jù)的總延遲時間均在100 ms以內(nèi)。

圖11以表格的形式展示了測試最后26批次數(shù)據(jù)的結(jié)果,包括每批數(shù)據(jù)的條數(shù)和提交時間,也就是包含多少條能耗數(shù)據(jù)、調(diào)度延遲時間、處理時間和總延遲時間。通過圖11的數(shù)據(jù),可以更進(jìn)一步佐證上面對測試數(shù)據(jù)的分析結(jié)果。從圖11可以看出每批數(shù)據(jù)平均包含146.88條能耗數(shù)據(jù);平臺調(diào)度延遲時間為0.34 ms;每批數(shù)據(jù)的平均處理時間為117.65 ms;平均總延遲時間為118 ms。所以,實時能耗分項計量的吞吐量為每秒處理1 248條記錄(146.88/117.65×1 000)。

圖11的數(shù)據(jù)是在系統(tǒng)度過了初始階段達(dá)到穩(wěn)定以后的數(shù)據(jù),結(jié)合圖10的整體情況,可以知道圖11的數(shù)據(jù)更能代表實時能耗分項計量系統(tǒng)的特性。

圖10 系統(tǒng)實時性測試結(jié)果

圖11 系統(tǒng)實時性詳細(xì)測試結(jié)果

3.3 測試結(jié)論

從上面的實驗結(jié)果可以看出,在實時能耗分項計量系統(tǒng)啟動以后,只需要處理完第一批數(shù)據(jù)以后,就能達(dá)到穩(wěn)定的運(yùn)行狀態(tài),大約3 s。平均實時能耗分項計量系統(tǒng)的吞吐量為每秒處理1 248條記錄,平臺調(diào)度延遲時間為0.34 ms;每批數(shù)據(jù)的平均處理時間為117.65 ms;平均總延遲時間為118 ms。所以,實時能耗分項計量系統(tǒng)具有很高的吞吐量,實時性很強(qiáng),并且系統(tǒng)數(shù)據(jù)處理速率很平穩(wěn)。

按照國家住建部分項計量規(guī)則要求,每塊分項計量儀表需要每15 min提交一次數(shù)據(jù)；而在15 min時間內(nèi),實時能耗分項計量系統(tǒng)可以處理超過100萬條(15×60×1 200)數(shù)據(jù)。也就是說,在現(xiàn)有的系統(tǒng)配置環(huán)境下,實時能耗分項計量系統(tǒng)可以支持100萬塊儀表。因為傳統(tǒng)的能耗分項計量系統(tǒng)需要先把數(shù)據(jù)寫入磁盤文件,然后再讀入進(jìn)行數(shù)據(jù)處理,并且沒有采用大數(shù)據(jù)并發(fā)處理技術(shù),所以每套系統(tǒng)能支持的分項計量儀表一般都在1 000塊左右,只適合于單個企事業(yè)單位的分項計量工作。實時能耗分項計量系統(tǒng)將處理能耗數(shù)據(jù)的能力提升了上千倍,完全可以滿足同一個城市的所有公共事業(yè)單位提供分項計量服務(wù)。

4 結(jié)語

本文提出了一種基于Spark Streaming和Apache Kafka模塊構(gòu)建的用于能耗分項計量的實時流式處理系統(tǒng),簡稱實時能耗分項計量系統(tǒng)。它能夠滿足能耗分項計量數(shù)據(jù)產(chǎn)生快、實時性強(qiáng)、數(shù)據(jù)量大的數(shù)據(jù)處理需求。與傳統(tǒng)數(shù)據(jù)處理架構(gòu)不同,實時能耗分項計量系統(tǒng)不僅提供離線數(shù)據(jù)的統(tǒng)計與分析,并且根據(jù)業(yè)務(wù)需求對數(shù)據(jù)進(jìn)行實時在線處理,在數(shù)據(jù)流動的過程中實時地捕捉異常信息并進(jìn)行處理,最終把結(jié)果保存或者分發(fā)給需要的組件。本文詳細(xì)描述了實時能耗分項計量系統(tǒng)的整體架構(gòu)和內(nèi)部結(jié)構(gòu),闡述了其主要特點,并通過實際測試證明了其強(qiáng)大的數(shù)據(jù)處理能力和實時性。

從功能方面來講,與傳統(tǒng)的能耗分項計量系統(tǒng)相比,實時能耗分項計量系統(tǒng)的最大優(yōu)點就是在支持離線能耗統(tǒng)計的同時,還可以支持實時在線數(shù)據(jù)處理和深度數(shù)據(jù)挖掘。比如,可以對能耗數(shù)據(jù)流進(jìn)行實時分析,發(fā)現(xiàn)能耗用量異常情況,及時報警給用戶,以便用能單位可以及時采取相應(yīng)措施,防止異常情況蔓延。再比如,實時能耗分項計量系統(tǒng)還可以實時統(tǒng)計各計量點能耗情況并實時展示給用戶,使用能單位及時掌握整體能耗的實時現(xiàn)狀。

從性能方面來講,本文提出的實時能耗分項計量系統(tǒng)進(jìn)行能耗數(shù)據(jù)處理的能力遠(yuǎn)遠(yuǎn)超過傳統(tǒng)的能耗分項計量系統(tǒng),能夠支持能耗數(shù)據(jù)采集點的個數(shù)高出上千倍。并且,實時能耗分項計量系統(tǒng)具有很強(qiáng)的擴(kuò)展能力,可以通過增加服務(wù)器和存儲設(shè)備來提高其總體處理能力,從而可以支持更多的能耗數(shù)據(jù)采集點。

總之,本文提出的實時能耗分項計量系統(tǒng)不論從性能方面、功能方面,還是從系統(tǒng)的可擴(kuò)展方面都遠(yuǎn)優(yōu)于傳統(tǒng)的能耗分項計量系統(tǒng)。本系統(tǒng)的第一版開發(fā)已經(jīng)完成,已經(jīng)在2016年開始在四川省進(jìn)行實地部署。此外,本文提出的實時流式數(shù)據(jù)處理系統(tǒng)還可以應(yīng)用于其他流式數(shù)據(jù)處理場合,比如股市走向分析、氣象數(shù)據(jù)測控、網(wǎng)站用戶行為分析和公路卡口過車數(shù)據(jù)分析等。

References)

[1] 清華大學(xué)建筑節(jié)能研究中心. 中國建筑節(jié)能年度發(fā)展研究報告2010[M]. 北京:中國建筑工業(yè)出版社, 2010:105-130.(Building energy conservation research center of tsinghua university. Annual Report of China Building Energy Conservation 2010[M]. Beijing:China Architecture and Building Press, 2010:105-130.)

[2] 魏慶芃. 大型公共建筑能耗分項計量實時監(jiān)測分析系統(tǒng)EMS-II的發(fā)展[J]. 建筑, 2009(3):34-37.(WEI Q P. Development of the detailed classification energy consumption measurement system for large public building EMS-II [J]. Construction and Architecture, 2009(3):34-37.)

[3] EMC-2000能源管理系統(tǒng)[EB/OL]. [2016- 09- 10]. http://www.hysine.cn/web/list/369/1.html.(EMC-2000 energy management system [EB/OL]. [2016- 09- 10]. http://www.hysine.cn/web/list/369/1.html.)

[4] 黃斌, 杜運(yùn)東, 曹雪華. 基于Acrel-5000的大型公共建筑能耗監(jiān)測系統(tǒng)設(shè)計與應(yīng)用[J]. 智能建筑電氣技術(shù), 2009, 3(5):47-50.(HUANG B, DU Y D, CAO X H. Design and application of large public building energy consumption monitoring system Acrel-5000[J]. Electrical Technology of Intelligent Building, 2009, 3(5):47-50.)

[5] 國家機(jī)關(guān)辦公建筑和大型公共建筑能耗監(jiān)測系統(tǒng)——分項計量數(shù)據(jù)采集技術(shù)導(dǎo)則[S]. 北京:中華人民共和國住房和城鄉(xiāng)建設(shè)部, 2008:1-25.(Government offices and large public buildings energy consumption monitoring system — the technical guidance for detailed classification energy data collection[S]. Beijing: Ministry of Housing and Urban-Rural Development of the People’s Republic of China, 2008:1-25.

[6] 國家機(jī)關(guān)辦公建筑和大型公共建筑能耗監(jiān)測系統(tǒng)——省、市級數(shù)據(jù)中心數(shù)據(jù)庫結(jié)構(gòu)文檔[S]. 北京:中華人民共和國住房和城鄉(xiāng)建設(shè)部, 2009:1-12.(Government offices and large public buildings energy consumption monitoring system — provincial and municipal data center database structure document[S]. Beijing: Ministry of Housing and Urban-Rural Development of the People’s Republic of China, 2009:1-12.)

[7] Spark programming guide[EB/OL]. [2016- 07- 27]. http://spark.apache. org /docs/latest/programming-guide.html.

[8] Spark streaming programming guide [EB/OL]. [2016- 07- 27]. https://sp ark.apache.org/docs/latest/programming-guide.html.

[9] 陸嘉恒. Hadoop實戰(zhàn)[M]. 北京:機(jī)械工業(yè)出版社, 2012:1-121.(LU J H. Hadoop in Action[M]. Beijing: China Machine Press, 2012:1-121.)

[10] KARAU H, KONWINSKI A, WENDELL P, et al. Spark快速大數(shù)據(jù)分析[M]. 王道遠(yuǎn), 譯. 北京: 人民郵電出版社, 2015: 161-185.(KARAU H, KONWINSKI A, WENDELL P, et al. Learning Spark: Lighting-Fast Data Analysis [M]. WANG D Y, translated. Beijing:Posts and Telecom Press, 2015: 161-185.)

[11] 夏俊鸞, 邵賽賽. Spark Streaming:大規(guī)模流式數(shù)據(jù)處理的新貴[J]. 程序員, 2014(2):44-48.(XIA J L, SHAO S S. Spark streaming: large-scale streaming data processing upstart[J]. Programmer, 2014(2):44-48.)

[12] Apache spark FAQ [EB/OL]. [2016- 08- 04]. https://spark.apache.org/faq.html.

[13] Apache Kafka: a high-throughput distributed messaging system [EB/OL]. [2016- 01- 09]. http://kafka.apache.org/documentation.html.

[14] KREPS J, NARKHED N, RAO J. Kafka: a distributed messaging system for log processing[C]// NetDB2011: Proceedings of the 6th International Workshop on Networking Meets Databases. New York: ACM, 2011: Article No. 12.

[15] 國家機(jī)關(guān)辦公建筑和大型公共建筑能耗監(jiān)測系統(tǒng)——數(shù)據(jù)上傳XML格式文檔[S]. 北京:中華人民共和國住房和城鄉(xiāng)建設(shè)部, 2009:55-59.(Government offices and large public buildings energy consumption monitoring system — XML format for data uploading [S]. Beijing: Ministry of Housing and Urban-Rural Development of the People’s Republic of China, 2009:55-59.)

[16] GEORGE L. HBase權(quán)威指南[M]. 代志遠(yuǎn), 劉佳, 蔣杰, 譯. 北京:人民郵電出版社, 2013:5-25.(GEORGE L. HBase: the Definitive Guide[M]. DAI Z Y, LIU J, JIANG J, translated. Beijing:Posts and Telecom Press, 2013:5-25.)

WU Zhixue, born in 1960, Ph. D., professor. His research interests include cloud computing, streaming data processing, data mining.

Real-time detailed classification energy consumption measurement system based on Spark Streaming

WU Zhixue1,2*

(1. Chengdu Wuzhou Handge Technology Limited, Chengdu Sichuan 611731, China;2. School of Information Security Engineering, Chengdu University of Information Technology, Chengdu Sichuan 610225, China)

Detailed classification energy consumption measurement can discover energy consuming issues more accurately, timely and effectively, which can form and implement the most effective energy-saving measures. Detailed classification energy measurement system needs to calculate energy consumption amounts at multiple time scales according to detailed classification coding. Not only does it need to complete the tasks timely, but also need to deal with data aggregating, data de-duplication and data joining operations. Due to the fast speed of the data being generated, the requirement of the data being processed in real-time, and the big size of the data volume, it is difficult to store the data to a database system first, and then to process the data afterwards. Therefore, the traditional data processing infrastructure cannot fulfil the requirements of detailed classification energy consumption measurement system. A new real-time detailed classification energy consumption measurement system based on Spark Streaming technologies was designed and implemented, the system infrastructure and the internal structure of the system were introduced in detail, and its real-time data processing capabilities were proved through experiments. Different from the traditional ways, the proposed system processes energy consumption data in real-time to capture any unusual behaviour timely; at the same time, it separates the data and calculates the consumption usages according to the detailed classification coding, and stores the results to a database system for offline analysis and data mining, which can effectively solve the previously mentioned problems encountered in the data processing process.

stream computing; detailed classification energy consumption measurement; Spark Streaming; Apache Kafka; big data

2016- 10- 10;

2016- 12- 21。

武志學(xué)(1960—),男,山西河津人,教授,博士,主要研究方向:云計算、流式數(shù)據(jù)處理、數(shù)據(jù)挖掘。

1001- 9081(2017)04- 0928- 08

10.11772/j.issn.1001- 9081.2017.04.0928

TP391

A