Spark Streaming框架下的氣象自動站數(shù)據(jù)實時處理系統(tǒng)

趙文芳，劉旭林

(1.北京市氣象信息中心，北京 100089； 2.北京市氣象探測中心，北京100176)(*通信作者電子郵箱lxulin@bjmb.gov.cn)

0 引言

目前，氣象自動站實現(xiàn)了分鐘加密觀測，具有產(chǎn)生快、實時性強(qiáng)、數(shù)據(jù)量大等特點，已經(jīng)成為監(jiān)視天氣變化、決策服務(wù)輔助支持的重要手段[1-3]，不僅在天氣預(yù)報、氣象服務(wù)中發(fā)揮重要作用，同時也為大城市防災(zāi)減災(zāi)、適應(yīng)氣候變化、環(huán)境評估等提供基礎(chǔ)的數(shù)據(jù)支撐。氣象業(yè)務(wù)現(xiàn)代化的進(jìn)展對自動站數(shù)據(jù)提出了越來越高的要求，中國氣象局在“十三五”規(guī)劃中明確提出了要實現(xiàn)自動站數(shù)據(jù)1 min內(nèi)到達(dá)預(yù)報員桌面的要求；北京市氣象局提出了要實現(xiàn)自動站數(shù)據(jù)在秒級甚至更短時間內(nèi)完成各類要素實時統(tǒng)計的要求，以便為災(zāi)害天氣的監(jiān)視提供及時準(zhǔn)確的氣象服務(wù)?；谧詣诱镜臄?shù)據(jù)服務(wù)面臨很大的挑戰(zhàn)，需要在秒級甚至更短時間內(nèi)完成高并發(fā)的處理存儲、統(tǒng)計和在線共享服務(wù)，因此，實現(xiàn)自動站數(shù)據(jù)的快速處理、有效存儲和及時服務(wù)，盡可能發(fā)揮它的應(yīng)用效益，顯得尤為重要。

Spark作為下一代大數(shù)據(jù)處理引擎，具有速度快、易用、通用等特點，在近幾年廣泛應(yīng)用于多個行業(yè)的并行數(shù)據(jù)分析，得到了很多領(lǐng)域和科研機(jī)構(gòu)的肯定。Spark Streaming是Spark提供的對實時數(shù)據(jù)進(jìn)行流式計算的組件,可以進(jìn)行大規(guī)模的流式數(shù)據(jù)處理，HBase是一個高可靠性、高性能、面向列的分布式數(shù)據(jù)庫，適合處理實時隨機(jī)讀寫超大規(guī)模的數(shù)據(jù)。將實時計算框架和分布式數(shù)據(jù)庫系統(tǒng)結(jié)合起來實現(xiàn)對海量數(shù)據(jù)的實時處理和高效檢索查詢，逐步成為了大數(shù)據(jù)技術(shù)在汽車、交通、氣象等行業(yè)應(yīng)用的新興研究領(lǐng)域[4-7]。例如，美國最權(quán)威的集汽車銷售和資訊服務(wù)為一體的網(wǎng)站Edmunds，結(jié)合Spark Streaming技術(shù)與HBase建立了實時儀表板系統(tǒng)，用于顯示用戶的活動信息、訪客ID、日志內(nèi)容和頁面瀏覽內(nèi)容等。考克斯汽車公司基于Spark Streaming技術(shù)創(chuàng)建實時儀表狀態(tài)系統(tǒng)，監(jiān)控道路狀況、汽車各項指標(biāo)、駕駛者的行為數(shù)據(jù)[8]。國家氣象信息中心也開展了基于HBase建立氣象地面分鐘數(shù)據(jù)分布式存儲系統(tǒng)的研究，以及Hadoop在氣象數(shù)據(jù)密集型領(lǐng)域的應(yīng)用實驗；廣東省氣象局開展了基于Hadoop實現(xiàn)風(fēng)暴追蹤算法和數(shù)值預(yù)報產(chǎn)品的服務(wù)應(yīng)用實驗等[9-16]。

本文結(jié)合北京市氣象局的氣象預(yù)報、決策氣象與公眾氣象服務(wù)等業(yè)務(wù)對自動站數(shù)據(jù)的需求，尤其是實況和統(tǒng)計的時效要求，重點研究了Spark Streaming框架下的自動站數(shù)據(jù)流處理技術(shù)和基于HBase的存儲管理技術(shù)，設(shè)計了并行的自動站流式入庫、實時溫度極值統(tǒng)計、風(fēng)要素極值統(tǒng)計和降水量累計等算法，實現(xiàn)了一種分布式的自動站數(shù)據(jù)流式處理，在Cloudera環(huán)境下建立了自動站實時處理系統(tǒng)。

1 業(yè)務(wù)現(xiàn)狀

目前，北京市氣象局主要依托全國綜合氣象信息共享系統(tǒng)(China Integrated Meteorological Information Sharing System， CIMISS)[17]實現(xiàn)京津冀自動站數(shù)據(jù)的處理、入庫存儲與管理，通過CIMISS的氣象數(shù)據(jù)統(tǒng)一服務(wù)接口(Meteorological Unified Service Interface Community， MUSIC)來獲取實時數(shù)據(jù)。CIMISS系統(tǒng)中自動站數(shù)據(jù)的全流程處理平均耗時不超過1 min，基本能滿足業(yè)務(wù)對實況的時效要求。然而，由于MUSIC接口沒有提供自動站要素的實時統(tǒng)計功能，因此只能通過編寫客戶端程序?qū)崿F(xiàn)，即遠(yuǎn)程調(diào)用接口服務(wù)獲取某時間段內(nèi)的要素值再進(jìn)行計算得出結(jié)果。這種做法存在一定弊端：1)由客戶端程序計算的要素統(tǒng)計值不能回寫到CIMISS數(shù)據(jù)庫中，無法實現(xiàn)統(tǒng)計功能的共享，造成很多不必要的重復(fù)性開發(fā)工作。2)由不同開發(fā)人員編寫的客戶端程序，可能由于算法差異導(dǎo)致統(tǒng)計結(jié)果不一致，從而降低數(shù)據(jù)的可用性。3)客戶端程序需要經(jīng)過多次迭代運(yùn)算才能得出最終結(jié)果，耗時較長，無法在秒級甚至更短時間內(nèi)實現(xiàn)，不能滿足用戶對自動站實時統(tǒng)計的時效需求，迫切需要改進(jìn)。

2 系統(tǒng)設(shè)計

2.1 系統(tǒng)技術(shù)框架

CDH(Cloudera Distribution Hadoop)是Cloudera公司發(fā)行的Apache Hadoop項目軟件包，里面包含了Hadoop和運(yùn)行在其之上的各類存儲計算框架，如Spark、Hive、HBase、Flume、Impala和Cloudera Search等。除了擁有開源Apache Hadoop的優(yōu)勢，CDH還具有以下獨特的優(yōu)點：1)使用Cloudera Manager可以實現(xiàn)CDH的自動安裝，簡化了Hadoop的安裝部署，避免了繁雜的多節(jié)點配置工作。2)對于Hadoop及其生態(tài)系統(tǒng)監(jiān)控都非常方便。Cloudera Manager提供一個基于Web的用戶界面，可以查看群集運(yùn)行狀況，修改CDH相關(guān)配置以及管理CDH各種服務(wù)。

為了滿足實際的業(yè)務(wù)需求，本文采用了Flume技術(shù)、Spark Streaming技術(shù)、CDH技術(shù)和HBase技術(shù)。其中，F(xiàn)lume技術(shù)完成自動站數(shù)據(jù)從觀測臺站到CDH平臺的實時發(fā)送；Spark Streaming技術(shù)實現(xiàn)對自動站數(shù)據(jù)的流式處理和要素統(tǒng)計的并行化處理，以提高運(yùn)算時效；HBase為系統(tǒng)提供數(shù)據(jù)存儲服務(wù)，CDH技術(shù)為系統(tǒng)提供平臺支撐和運(yùn)行環(huán)境。

Flume是分布式的，可以同步處理到達(dá)的多個文件。Spark Streaming將輸入數(shù)據(jù)流以時間片(秒級)為單位進(jìn)行拆分，然后以類似批處理的方式處理每個時間片數(shù)據(jù)，使用基于內(nèi)存的Spark作為執(zhí)行引擎，具有高效和容錯的特性，并達(dá)到秒級延遲。利用Spark Streaming進(jìn)行自動站的數(shù)據(jù)處理，在幾十毫秒內(nèi)就能完成自動站入庫和要素統(tǒng)計查詢?nèi)蝿?wù)，完全能滿足氣象自動站數(shù)據(jù)1 min內(nèi)到達(dá)預(yù)報員桌面的要求，因此，將Flume和Spark Streaming結(jié)合起來，可以實現(xiàn)自動站數(shù)據(jù)的實時高效處理。

本文的研究和實驗均基于CDH的分布式計算框架而開展。

2.2 體系結(jié)構(gòu)設(shè)計

系統(tǒng)體系結(jié)構(gòu)設(shè)計如圖1所示，主要包括4個層次：表現(xiàn)層、邏輯層、服務(wù)層和數(shù)據(jù)層,是一種典型的多層體系結(jié)構(gòu)。表現(xiàn)層也是用戶層，主要指客戶端，可以是瀏覽器或者業(yè)務(wù)系統(tǒng)等。邏輯層對應(yīng)系統(tǒng)的功能，包括數(shù)據(jù)查詢檢索、數(shù)據(jù)入庫和要素統(tǒng)計等，由運(yùn)行在Spark Streaming計算框架服務(wù)之上的功能模塊實現(xiàn)。服務(wù)層是運(yùn)行于在CDH平臺上的各種服務(wù)，這里主要包括實現(xiàn)HBase訪問接口的REST (REpresentational Sate Transfer)網(wǎng)關(guān)(Gateway)服務(wù)和Thrift網(wǎng)關(guān)(Gateway)服務(wù)以及Spark Streaming計算框架服務(wù)。數(shù)據(jù)層指應(yīng)用數(shù)據(jù)的存儲層，該層利用HBase存儲了氣象自動站實況數(shù)據(jù)和多種要素統(tǒng)計數(shù)據(jù)。

圖1 系統(tǒng)體系結(jié)構(gòu)

2.3 功能設(shè)計

從數(shù)據(jù)的處理流程來講，系統(tǒng)包括數(shù)據(jù)收集、數(shù)據(jù)入庫、要素統(tǒng)計、數(shù)據(jù)查詢檢索、基于MUSIC的HBase數(shù)據(jù)接口訪問五大業(yè)務(wù)功能。

1)數(shù)據(jù)收集。利用CIMISS的數(shù)據(jù)收集與分發(fā)系統(tǒng)(Collecting and Transferring System, CTS)系統(tǒng)將觀測數(shù)據(jù)發(fā)送到Flume的監(jiān)控目錄，F(xiàn)lume將收到的數(shù)據(jù)實時發(fā)送到Spark Streaming的監(jiān)控輸入目錄。Flume是分布式的，可以同步處理到達(dá)的多個文件，同時它也提供了許多可調(diào)的故障恢復(fù)和容錯機(jī)制，當(dāng)某個節(jié)點出現(xiàn)故障時，數(shù)據(jù)能夠被傳送到其他節(jié)點上而不會丟失，從而保證數(shù)據(jù)的完整性。

2)數(shù)據(jù)入庫。支持國家級自動站和區(qū)域自動站格式的解析讀取，為自動站數(shù)據(jù)創(chuàng)建自定義的數(shù)據(jù)結(jié)構(gòu)體，通過Map映射將原始的文本氣象數(shù)據(jù)轉(zhuǎn)為包含數(shù)據(jù)結(jié)構(gòu)體的離散流(Discretized Stream， DStream)對象序列，從而寫入HBase。

3)要素統(tǒng)計。支持自動站常規(guī)要素的各類實時統(tǒng)計，主要包括逐3 h、6 h、12 h、24 h的溫度極值和風(fēng)極值統(tǒng)計以及逐5 min、10 min、3 h、6 h、12 h、24 h累計降水量統(tǒng)計。利用Spark對數(shù)據(jù)進(jìn)行轉(zhuǎn)換和聚合，通過多次的迭代計算實時實現(xiàn)每個站點的統(tǒng)計結(jié)果，寫入HBase的相關(guān)數(shù)據(jù)表中。

4)數(shù)據(jù)的查詢檢索。為自動站實況和統(tǒng)計提供各種條件的查詢檢索功能。使用Solr(Search On Lucene Replication)對查詢常用字段建立輔助索引，通過Solr多條件查詢快速獲得符合過濾條件的Rowkey，通過指定Rowkey在HBase中查詢到符合條件的結(jié)果。

5)基于MUSIC的HBase數(shù)據(jù)接口訪問。HBase提供了兩種接口方式獲取自動站數(shù)據(jù)，包括Thrift網(wǎng)關(guān)(Gateway)和REST網(wǎng)關(guān)(Gateway)服務(wù)。其中，Thrift網(wǎng)關(guān)(Gateway)服務(wù)利用Thrift序列化技術(shù)用以支持C++、PHP(Hypertext Preprocessor)、Python等多種語言對HBase的訪問；REST網(wǎng)關(guān)(Gateway)服務(wù)支持超文本傳輸協(xié)議(HyperText Transfer Protocol, HTTP)的應(yīng)用編程接口(Application Programming Interface, API)方式訪問HBase。系統(tǒng)通過拓展MUSIC接口功能，增加支持REST網(wǎng)關(guān)(Gateway)服務(wù)和Thrift網(wǎng)關(guān)(Gateway)服務(wù)的接口函數(shù)，實現(xiàn)了MUSIC接口服務(wù)對HBase的數(shù)據(jù)訪問。

3 系統(tǒng)關(guān)鍵技術(shù)

3.1 基于列模式設(shè)計自動站表結(jié)構(gòu)

3.1.1 Rowkey和列族的設(shè)計

Rowkey是HBase表的主鍵，設(shè)計合理的Rowkey有助于提高HBase的查詢檢索速度。自動站數(shù)據(jù)是基于時間序列的，它的查詢和統(tǒng)計都和觀測時間相關(guān)，必須將觀測時間存入到Rowkey中，但是，含有觀測時間的Rowkey會按時間戳的方式單調(diào)遞增，很容易引起單區(qū)熱點問題，從而不能將集群的整體性能發(fā)揮出來。為了避免上述問題的出現(xiàn)，該系統(tǒng)將行主鍵中的站號利用MD5(Message-Digest Algorithm 5)方法散列化，以便將所有的數(shù)據(jù)散列到不同的Region上。自動站表模式設(shè)計如表1所示，由行主鍵加兩個列族構(gòu)成。行主鍵的設(shè)計采用MD5(站號)+觀測時間組合的方式，兩個列族分別存放自動站觀測要素和自動站要素統(tǒng)計值。例如，每分鐘的溫、壓、濕、風(fēng)等要素觀測值對應(yīng)于列族aws_data，而自動站的溫度統(tǒng)計、風(fēng)統(tǒng)計、降水量統(tǒng)計值則對應(yīng)于列族aws_stat。

表1 自動站表結(jié)構(gòu)

3.1.2 利用MD5實現(xiàn)對行主鍵Rowkey散列化

MD5是在20世紀(jì)90年代初由麻省理工學(xué)院(Massachusetts Institute of Technology, MIT)的計算科學(xué)實驗室和RSA(Rivest-Shamir-Adleman) Data Security Incorporation開發(fā)的，經(jīng)MD2(Message-Digest Algorithm 2)、MD3(Message-Digest Algorithm 3)和MD4(Message-Digest Algorithm 4)發(fā)展而來的，在計算機(jī)安全領(lǐng)域廣泛使用的一種散列函數(shù)。Message-Digest泛指字節(jié)串的哈希變換，就是將一個任意長度的字節(jié)串變換成一定長的整數(shù)(128位)，成為一個不可逆的字符串，這樣即使用戶看到源程序和算法描述，也無法將一個MD5值變換回原始的字符串，從而確保了數(shù)據(jù)的安全性[18-19]。

本文在設(shè)計自動站數(shù)據(jù)表的Rowkey時，將字符串(String)類型的觀測時間轉(zhuǎn)化為時間戳存為長整(Long)型，對字符串(String)類型長度為5的站號利用MD5加密，加密后的站號+Long型的觀測時間，MD5(站號)+時間(YYYYMMDDHHMM)作為Rowkey。以下是創(chuàng)建Rowkey的Java部分代碼。

String stn=cells[0];

String otime=cells[1];

String otime_long=String.format("%s%s%s%s%s",otime.substring(0,4),otime.substring(5,7),otime.substring(8,10),otime.substring(11,13),otime.substring(14,16));

Long currTime=Long.parseLong(otime_long);

byte[] userHash=Md5Utils.md5sum(stn);

byte[] timestamp=Bytes.toBytes(-1*currTime);

byte[] Rowkey_b=new byte[Md5Utils.MD5_LENGTH+timestamp.length];

int offset=0;

offset=Bytes.putBytes(Rowkey_b,offset,userHash,0,userHash.length);

Bytes.putBytes(Rowkey_b,offset,timestamp,0,timestamp.length);

Put put=new Put(Rowkey_b);

3.2 自動站流式入庫和要素統(tǒng)計算法設(shè)計

3.2.1 流式入庫算法設(shè)計

自動站流式入庫算法用于將原始文本數(shù)據(jù)寫入HBase表。自動站數(shù)據(jù)包括自動站號、觀測時間、溫度、降水、濕度、風(fēng)等要素觀測值，以字符串方式存儲，每一行代表一個自動站記錄。Spark Streaming處理的數(shù)據(jù)流DStream由一系列RDD(Resilient Distributed Datasets)組成。Spark Streaming的編程就是利用自帶的接口來對DStream進(jìn)行操作，因此，參照這種編程思想，該算法的關(guān)鍵在于自動站數(shù)據(jù)從原始文本格式到DStream的轉(zhuǎn)換，需要定義一個類對象用于存儲自動站要素觀測值，然后生成DStream<類對象>數(shù)據(jù)集，通過map函數(shù)實現(xiàn)原始數(shù)據(jù)到DStream的映射轉(zhuǎn)換。具體的算法步驟描述如下：

1)將原始數(shù)據(jù)讀入并存為DStream 類型，即每個自動站的原始數(shù)據(jù)表示為一個String類型的DStream。

2)創(chuàng)建一個類對象aws_access，用于存放自動站的相關(guān)信息與數(shù)據(jù)，包括臺站基本信息、觀測時間、觀測要素值等。定義成員變量和成員函數(shù)，支持要素觀測值的獲取和設(shè)置以及對原始文本數(shù)據(jù)的逐行解析。

3)自定義一個繼承接口函數(shù)，用于解析原始數(shù)據(jù)并轉(zhuǎn)化為aws_access對象。

4)對DStream 進(jìn)行操作轉(zhuǎn)換。調(diào)用JavaDStream的Map函數(shù)將DStream 類型轉(zhuǎn)為DStream 。

5)對數(shù)據(jù)集DStream 調(diào)用Foreach/函數(shù)，新建Put對象并將自動站數(shù)據(jù)賦值給Put使之初始化。

6)提交Put對象完成數(shù)據(jù)寫入HBase。

自動站數(shù)據(jù)寫入HBase算法流程如圖2所示。

圖2 自動站數(shù)據(jù)寫入HBase算法流程

3.2.2 要素統(tǒng)計算法設(shè)計

設(shè)計合理的鍵值對(key，value)，實現(xiàn)從實際數(shù)據(jù)到鍵值對的映射，可以很方便地實現(xiàn)迭代計算。

自動站逐5 min累計降水量算法步驟描述如下：

1)建立一個選擇條件，用于從HBase中獲取1 h內(nèi)的所有站的分鐘降水?dāng)?shù)據(jù)。

2)從HBase中獲取相應(yīng)結(jié)果并轉(zhuǎn)為JavaRDD對象。

3)調(diào)用鍵值對轉(zhuǎn)換接口函數(shù)MaptoPair將JavaRDD對象轉(zhuǎn)為JavaPairRDD對象(key，value)；選擇站號+觀測時間組合方式作為key，value為每分鐘降水，如果該分鐘降水值缺測，則設(shè)置為0。觀測時間為5的倍數(shù),單位為分鐘，即每小時的05分、10分……。以站號為54511、觀測時間2016年4月20日08時05分為例，將觀測時間是01-05分(即08:00—08:05)的數(shù)據(jù)，都轉(zhuǎn)為觀測時間為2016年4月20日08時05分的(key，value)鍵值對。

4)對JavaPaieRDD對象運(yùn)用groupByKey函數(shù)和Reduce函數(shù)，得出統(tǒng)計結(jié)果。

5)創(chuàng)建Put對象，根據(jù)統(tǒng)計結(jié)果對Put對象進(jìn)行初始化，最后提交給HBase寫入數(shù)據(jù)表相應(yīng)的列從而完成統(tǒng)計過程。

逐10 min、3 h、6 h、12 h、24 h累計降水量算法與逐5 min累計降水量算法類似，溫度和風(fēng)要素極值統(tǒng)計與降水量算法步驟也類似，只是調(diào)用的迭代函數(shù)不一樣。

3.2.3 動態(tài)資源池應(yīng)用

動態(tài)資源池用于給YARN(Yet Another Resource Negotiator)或 Impala應(yīng)用指定資源配置和策略。系統(tǒng)利用Cloudera Manager的Web管理功能，添加一個動態(tài)資源池用來運(yùn)行Spark Streaming提交的作業(yè)任務(wù)，在YARN頁面的控制面板配置權(quán)重、虛擬內(nèi)核、內(nèi)存大小以及正在運(yùn)行的應(yīng)用程序最大數(shù)量等參數(shù)。通過調(diào)整這些參數(shù)和分析對比運(yùn)行Spark Streaming作業(yè)的耗時，找出作業(yè)運(yùn)行的最佳參數(shù)設(shè)置方案，實現(xiàn)Spark Streaming運(yùn)行調(diào)優(yōu)。

4 實驗與分析

為了檢驗氣象自動站實時處理系統(tǒng)進(jìn)行數(shù)據(jù)入庫處理和實時統(tǒng)計的能力，實際部署了一套自動站實時處理系統(tǒng)，并進(jìn)行了一系列的測試。

不同的氣象業(yè)務(wù)場景對氣象自動站資料的關(guān)注側(cè)重點有所不同。在實時天氣預(yù)報和氣象服務(wù)中，預(yù)報員對氣象自動站資料的關(guān)注重點是數(shù)據(jù)時效性，即分鐘、小時和日數(shù)據(jù)能否及時到達(dá)預(yù)報員桌面以及實現(xiàn)要素的實時快速統(tǒng)計；在天氣復(fù)盤推演、氣候極值統(tǒng)計和天氣過程分析等領(lǐng)域，預(yù)報員對氣象自動站資料的關(guān)注重點是時間序列較長的30 d和90 d數(shù)據(jù)的完整性和一致性。為了滿足這些需求，自動站資料至少需要保存1～3個月。分別選擇1 min、1 h、1 d、30 d、90 d的京津冀自動站分鐘數(shù)據(jù)作為5個測試數(shù)據(jù)集，依次編號為數(shù)據(jù)集1、數(shù)據(jù)集2、數(shù)據(jù)集3、數(shù)據(jù)集4、數(shù)據(jù)集5。京津冀總共3 802個自動站，一個站一個觀測時次產(chǎn)生的數(shù)據(jù)大約4 kb。

4.1 系統(tǒng)運(yùn)行環(huán)境

本文基于Apache Flume 1.5、Spark1.5、JDK(Java Development Kit) 1.7、HBase1.04實現(xiàn)上述系統(tǒng)原型,選擇Cloudera 5.5作為系統(tǒng)的運(yùn)行環(huán)境，采用YARN模式作為系統(tǒng)的運(yùn)行方式。其中，Cloudera集群由10臺實體服務(wù)器構(gòu)成，每臺服務(wù)器的配置為32核CPU，128 GB內(nèi)存，4 TB磁盤。

在Cloudera環(huán)境下，需要將YARN、HBase、Spark、Solr服務(wù)合理地部署在不同的節(jié)點上，以便充分利用集群資源給系統(tǒng)運(yùn)行提供更好的性能保障。選擇運(yùn)行Cloudera Manager服務(wù)的節(jié)點同時作為分布式文件系統(tǒng)(Hadoop Distributed File System, HDFS)的名字空間節(jié)點(namenode)，并運(yùn)行Spark Master、HMaster Server和YARN資源管理(ResourceManager) 服務(wù)，具體如圖3所示。

圖3 Cloudera服務(wù)部署

4.2 不同Rowkey對寫入HBase操作的性能影響測試

測試主要考察在數(shù)據(jù)量以及其他系統(tǒng)配置不變的情況下，通過設(shè)計不同的Rowkey而引起的HBase寫入耗時的變化，選擇本地模式運(yùn)行程序。

進(jìn)行兩組實驗，對編號為1～3的數(shù)據(jù)集進(jìn)行20次入庫測試，取平均耗時作為結(jié)果。當(dāng)設(shè)計Rowkey為時間(yyyymmddhhMM)+站號的方式時，3個數(shù)據(jù)集的平均入庫耗時分別為14 s、144 s、498 s；當(dāng)設(shè)計Rowkey為MD5(站號)+時間(yyyymmddhhMM)的方式時，3個數(shù)據(jù)集的平均入庫耗時分別為4 s、24 s、121 s。結(jié)果如圖4。

由圖4可以看出，如果Rowkey按照時間戳的方式遞增而且首字段直接是時間信息，所有的數(shù)據(jù)都會集中在HBase的一個分區(qū)服務(wù)(RegionServer)上，造成數(shù)據(jù)檢索時的熱點問題(hot splot)，導(dǎo)致查詢效率下降。由此可見，對于時間序列的數(shù)據(jù)集，Rowkey的設(shè)計對HBase的性能影響比較大。

圖4 HBase寫入性能隨Rowkey變化的測試結(jié)果

4.3 HBase寫入時效隨Spark運(yùn)行參數(shù)變化測試

進(jìn)行實時場景和批量數(shù)據(jù)寫入場景兩種測試，選擇Spark on YARN運(yùn)行方式。在進(jìn)行測試之前，先調(diào)整YARN服務(wù)的可用資源。在10個節(jié)點的集群中，每個節(jié)點有32個core以及128 GB的內(nèi)存；其中，1個節(jié)點運(yùn)行YARN資源管理(Resource Manager)服務(wù)，9個節(jié)點運(yùn)行YARN節(jié)點管理(NodeManager)服務(wù)；考慮到每個節(jié)點上操作系統(tǒng)、Hadoop的Daemon進(jìn)程以及其他組件進(jìn)程的運(yùn)行也需要一定資源，在該測試中分配給YARN 75%的資源，即每個節(jié)點上NodeManage可用資源為96 GB內(nèi)存和24個core。

實時場景的測試使用數(shù)據(jù)集1和數(shù)據(jù)集2作為樣本數(shù)據(jù)。啟動Spark Streaming應(yīng)用進(jìn)程，結(jié)合CTS和Flume將數(shù)據(jù)發(fā)送到Spark Streaming的監(jiān)控目錄下，通過Web頁面記錄每次作業(yè)的耗時。對每個數(shù)據(jù)集進(jìn)行20次測試，取平均耗時作為測試結(jié)果。測試結(jié)果表明，數(shù)據(jù)集1入庫平均耗時為30.64 ms，數(shù)據(jù)集2入庫平均耗時為6.37 s，完全能滿足實時業(yè)務(wù)的需求。

批量數(shù)據(jù)寫入HBase場景測試的主要目的是考察在數(shù)據(jù)量、任務(wù)提交方式及其他配置不變的情況下，通過改變num-executors、executor-cores和executor-memory三個運(yùn)行參數(shù)而引起的時效變化。數(shù)據(jù)集3包括17 551個文件，數(shù)據(jù)總量389 MB，數(shù)據(jù)集4包括54萬個文件，數(shù)據(jù)總量11.86 GB，數(shù)據(jù)集5包括159萬多個文件，數(shù)據(jù)總量34.56 GB。考慮到數(shù)據(jù)集4和數(shù)據(jù)集5主要應(yīng)用在非實時業(yè)務(wù)場景，而數(shù)據(jù)集3主要應(yīng)用在實時業(yè)務(wù)場景，選擇對數(shù)據(jù)集3進(jìn)行9組實驗，詳細(xì)的參數(shù)配置如表2。

在實驗中，num-executors從10逐漸增大到90。由圖5可知，數(shù)據(jù)寫入HBase的耗時隨num-executors的增加呈下降趨勢。當(dāng)num-executors取值從[20,30)，[30,40)，[40,50)時，耗時并沒有明顯的變化。在num-executors分別設(shè)置為[60,70)，[70,80)，[80,90)的三組實驗中，啟用spark.default.parallelism參數(shù)，程序運(yùn)行效率得到提升。當(dāng)num-executors為70、executor-cores為3、executor-memory為4 GB、spark.default.parallelism為600時，整體性能達(dá)到最優(yōu)。spark.default.parallelism參數(shù)用于設(shè)置每個stage的默認(rèn)task數(shù)量。前5組實驗沒有啟用這個參數(shù)，而Spark默認(rèn)設(shè)置task為幾十個，導(dǎo)致60%至70%的Executor進(jìn)程沒有task執(zhí)行，因此盡管Executor參數(shù)在增大，但程序的耗時基本沒有變化。由此可見，這個參數(shù)如果不設(shè)置或者設(shè)置不當(dāng)會直接影響Spark作業(yè)性能。

表2 Spark Streaming作業(yè)運(yùn)行參數(shù)配置及耗時

圖5 耗時隨num-executors參數(shù)變化測試結(jié)果

對數(shù)據(jù)集4和數(shù)據(jù)集5進(jìn)行寫入HBase測試。啟用參數(shù)spark.default.parallelism并設(shè)置為1 000，spark運(yùn)行參數(shù)按照表7中序號為7的組合參數(shù)，數(shù)據(jù)4和數(shù)據(jù)集5耗時分別為161 s和249 s；相比現(xiàn)有業(yè)務(wù)環(huán)境下，數(shù)據(jù)集4批量導(dǎo)入關(guān)系型數(shù)據(jù)庫耗時196 min，數(shù)據(jù)集5批量導(dǎo)入關(guān)系型數(shù)據(jù)庫耗時578 min，性能分別提升了73倍和139倍。

從對數(shù)據(jù)集3～5的測試結(jié)果來看，Spark作業(yè)性能與運(yùn)行參數(shù)、平臺分配給YARN的可用資源、spark.default.parallelism等諸多因素有關(guān)，在其他參數(shù)不變的情況下，性能與num-executors參數(shù)也不是簡單的線性關(guān)系，因此，在實際運(yùn)行中，需要統(tǒng)籌考慮這些參數(shù)的影響，選擇一個最優(yōu)的參數(shù)組合。此外，針對大量小文件的批量導(dǎo)入HBase，除了考慮spark運(yùn)行參數(shù)，還可以先對原始文件進(jìn)行合并和sequence序列化處理；當(dāng)氣象自動站小文件累計容量達(dá)到40 GB以上，可以考慮增加Cloudera集群節(jié)點，通過擴(kuò)充硬件資源來提升效率。

4.4 不同查詢條件下的HBase查詢性能測試

進(jìn)行四組查詢測試，每組測試運(yùn)行20次，取其平均耗時，結(jié)果如表3。從表3可以看出，當(dāng)查詢只以Rowkey作為唯一條件時候，不借助Solr索引查詢響應(yīng)比較快，達(dá)到毫秒級。當(dāng)查詢需要根據(jù)時間設(shè)置HBase的start key和end key時候，不借助Solr索引查詢達(dá)到秒級，而借助索引后能達(dá)到毫秒級。

4.5 要素統(tǒng)計性能測試

對數(shù)據(jù)集進(jìn)行逐日和逐時溫度、風(fēng)、降水量要素的統(tǒng)計計算，耗時結(jié)果如表4。不同類型統(tǒng)計運(yùn)行耗時有一定差別，但都到達(dá)了秒級甚至毫秒級。然而，在現(xiàn)有業(yè)務(wù)平臺中完成一個月逐日的要素統(tǒng)計需要14～16 min才能完成，由此可見利用Spark Streaming來實現(xiàn)自動站要素統(tǒng)計是完全可行的。

4.6 系統(tǒng)的業(yè)務(wù)應(yīng)用案例

該系統(tǒng)投入業(yè)務(wù)化運(yùn)行后，全面支持了北京市氣象局的自動站數(shù)據(jù)基礎(chǔ)業(yè)務(wù)，為預(yù)報員和其他氣象工作人員提供了精準(zhǔn)的自動站實況和要素統(tǒng)計數(shù)據(jù)。尤其是自動站逐5 min累計雨量的實時統(tǒng)計功能，為各區(qū)局的氣象決策服務(wù)提供了十分方便的數(shù)據(jù)服務(wù)，同時提升了數(shù)據(jù)的綜合服務(wù)能力。此外，基于Thrift網(wǎng)關(guān)方式的數(shù)據(jù)接口服務(wù)已經(jīng)成功為市局多個業(yè)務(wù)系統(tǒng)提供自動站要素的統(tǒng)計數(shù)據(jù)支撐；在2016年7月24日和8月12日二次降水天氣過程中，以REST網(wǎng)關(guān)方式訪問系統(tǒng)數(shù)據(jù)的用戶總數(shù)達(dá)到556，由此可見，系統(tǒng)在市局和各區(qū)局得到了推廣應(yīng)用，贏得用戶一致好評。

表3 查詢用例

表4 自動站要素統(tǒng)計耗時

5 結(jié)語

本文基于Cloudera的CDH搭建了大數(shù)據(jù)平臺，將Flume和Spark Streaming技術(shù)相結(jié)合，研發(fā)了基于Spark Streaming的氣象自動站實時處理系統(tǒng)，可以快速實現(xiàn)自動站的要素統(tǒng)計和流式入庫。與傳統(tǒng)的處理方式相比，基于流式計算框架實現(xiàn)氣象數(shù)據(jù)的入庫處理和統(tǒng)計效率更高，處理流程也簡單?？梢?，該系統(tǒng)對提升自動站數(shù)據(jù)業(yè)務(wù)的綜合應(yīng)用能力具有重要意義，完全具備可行性和適用性。

另外，通過拓展MUSIC接口實現(xiàn)對HBase的數(shù)據(jù)訪問，為氣象業(yè)務(wù)系統(tǒng)提供自動站數(shù)據(jù)的實時要素統(tǒng)計查詢服務(wù)，既滿足了業(yè)務(wù)需求，同時也克服了CIMISS要素統(tǒng)計功能的不足。此外，利用HBase存儲自動站數(shù)據(jù)，可以作為CIMISS環(huán)境下自動站數(shù)據(jù)存儲的實時備份，為氣象業(yè)務(wù)提供更可靠的數(shù)據(jù)保障服務(wù)。這項工作為深入推進(jìn)CIMISS本地化開發(fā)和應(yīng)用提供了良好的借鑒。

在后續(xù)的研究與應(yīng)用工作中，將進(jìn)一步關(guān)注MUSIC的發(fā)展，研發(fā)更多的接口功能函數(shù)，通過MUSIC平臺發(fā)布出來，為更多的業(yè)務(wù)系統(tǒng)和用戶提供自動站的實時要素統(tǒng)計數(shù)據(jù)服務(wù)，更好發(fā)揮自動站數(shù)據(jù)的應(yīng)用效益。此外，還將繼續(xù)開發(fā)更多氣象資料處理和統(tǒng)計算法，在Spark Streaming下進(jìn)行算法的實現(xiàn)和實驗。

References)

[1] 田蘭,金石聲,李波,等.基于XML和正則表達(dá)式的氣象數(shù)據(jù)處理系統(tǒng)[J].計算機(jī)科學(xué),2013,40(11A):432-435.(TIAN L, JIN S S, LI B, et al. Processing system of meteorological data based on XML and regular expression [J]. Computer Science, 2013, 40(11A): 432-435.)

[2] 李峰,秦世廣,周薇,等.綜合氣象觀測運(yùn)行監(jiān)控業(yè)務(wù)及系統(tǒng)升級設(shè)計[J].氣象科技,2014,42(4):539-544.(LI F, QIN S G, ZHOU W, et al. Upgrading design of integrated atmospheric observing monitoring operation and system platform [J]. Meteorological Science and Technology, 2014, 42(4): 539-544.)

[3] 錢崢,曹艷艷,趙科科,等.私有云在市級氣象業(yè)務(wù)平臺的實現(xiàn)與應(yīng)用[J].氣象科技,2014,42(4):641-646.(QIAN Z, CHAO Y Y, ZHAO K K, et al. Implementation and application of private cloud in municipal-level meteorological operation platform [J]. Meteorological Science and Technology, 2014, 42(4): 641-646.)

[4] ZHAO S, YANG X, LI X, et al. A Hadoop-based visualization and diagnosis framework for earth science data [C]// Proceedings of the 2015 IEEE International Conference on Big Data. Piscataway, NJ: IEEE, 2015: 1972-1977.

[5] DUFFY D Q, SCHNASE J L, THOMPSON J H, et al. Preliminary evaluation of MapReduce for high-performance climate data analysis [EB/OL]. [2016- 04- 08]. https://ntrs.nasa.gov/archive/nasa/casi.ntrs.nasa.gov/20120009187.pdf.

[6] KARUN A K, CHITHARANJAN K. A review on Hadoop-HDFS infrastructure extensions [C]// Proceedings of the 2013 IEEE Conference on Information and Communication Technologies. Piscataway, NJ: IEEE,2013:132-137.

[7] VORA M N. Hadoop-HBase for large-scale data[C]// Proceedings of the 2011 International Conference on Computer Science and Network Technology. Piscataway, NJ: IEEE,2011: 601-605.

[8] 曾沁,李永生.基于分布式計算框架的風(fēng)暴三維追蹤方法[J].計算機(jī)應(yīng)用,2017,37(4):941-944.(ZENG Q, LI Y S. Three dimensional storm tracking method based on distributed computing architecture [J]. Journal of Computer Applications, 2017, 37(4): 941-944.)

[9] 李英俊,韓雷.基于三維雷達(dá)圖像數(shù)據(jù)的風(fēng)暴體追蹤算法研究[J].計算機(jī)應(yīng)用,2008,28(4):1078-1080.(LI Y J, HAN L. Storm tracking algorithm development based on the three-dimensional radar image data [J]. Journal of Computer Applications, 2008, 28(4): 1078-1080.)

[10] 鄭芳,許先斌,向冬冬,等.基于GPU的GRAPES數(shù)值預(yù)報系統(tǒng)中RRTM模塊的并行化研究[J].計算機(jī)科學(xué),2012,39(6A):370-374.(DENG F, XU X B, XIANG D D, et al. GPU-based parallel researches on RRTM module of GRAPES numerical prediction system [J]. Computer Science, 2012, 39(6A): 370-374.)

[11] 吳石磊,安虹,李小強(qiáng),等.組網(wǎng)雷達(dá)估測降水系統(tǒng)并行化方案的設(shè)計與實現(xiàn)[J].計算機(jī)科學(xué),2012,39(3):271-275.(WU S L, AN H, LI X Q, et al. Parallel program design and implementation on precipitation program of networking weather radar system [J]. Computer Science, 2012, 39(3): 271-275.)

[12] 楊潤芝,沈文海,肖衛(wèi)青,等.基于MapReduce計算模型的氣象資料處理調(diào)優(yōu)試驗[J].應(yīng)用氣象學(xué)報,2014,25(5):618-627.(YANG R Z, SHEN W H, XIAO W Q, et al. A set of MapReduce tuning experiments based on meteorological operations [J].Journal of Applied Meteorological Science, 2014, 25(5): 618-627.)

[13] 陳東輝,曾樂,梁中軍,等.基于HBase的氣象地面分鐘數(shù)據(jù)分布式存儲系統(tǒng)[J].計算機(jī)應(yīng)用,2014,34(9):2617-2621.(CHEN D H, ZENG L, LIANG Z J, et al. HBase-based distributed storage system for meteorological ground minute data [J]. Journal of Computer Applications, 2014, 34(9): 2617-2621.)

[14] 薛勝軍,劉寅.基于Hadoop的氣象信息數(shù)據(jù)倉庫建立與測試[J].計算機(jī)測量與控制,2012,20(4):926-929.(XUE S J, LIU Y. Establishment and test of meteorological data warehouse based on Hadoop [J]. Computer Measurement and Control, 2012, 20(4): 926-929.)

[15] 薛勝軍,周天波,周天杰.基于Hadoop的氣象云儲存與數(shù)據(jù)處理應(yīng)用淺析[J].數(shù)字技術(shù)與應(yīng)用,2012,15(5):82-84.(XUE S J, ZHOU T B, ZHOU T J. Analysis of meteorological cloud storage and data processing based on Hadoop [J]. Digital Technology & Application, 2012,15(5): 82-84.)

[16] 楊鋒,吳華瑞,朱華吉,等.基于Hadoop的海量農(nóng)業(yè)數(shù)據(jù)資源管理平臺[J].計算機(jī)工程,2011,37(12):222-224.(YANG F,WU H R, ZHU H J, et al. Massive agricultural data resource management platform based on Hadoop [J]. Computer Engineering, 2011, 37(12): 222-224.)

[17] 熊安元,趙芳,王穎,等.全國綜合氣象信息共享系統(tǒng)的設(shè)計與實現(xiàn)[J].應(yīng)用氣象學(xué)報,2015,26(4):500-513.(XIONG A Y, ZHAO F, WANG Y, et al. Design and implementation of China integrated meteorological information sharing system [J]. Journal of Applied Meteorological Science, 2015, 26(4): 500-513.)

[18] BHARDWAJ A,VANRAJ, KUMAR A, et al. Big data emerging technologies: a CaseStudy with analyzing twitter data using apache hive [C]// Proceedings of the 2015 2nd International Conference on Recent Advances in Engineering & Computational Sciences. Piscataway, NJ: IEEE, 2015: 1-6.

[19] 王金柱,李元誠.MD5算法在J2EE平臺下用戶管理系統(tǒng)中的應(yīng)用[J].計算機(jī)工程與設(shè)計,2008,29(18):4728-4764.(WANG J Z, LI Y C. Application of MD5 algorithm based on J2EE in user management system [J]. Computer Engineering and Design, 2008, 29(18): 4728-4764.

This work is partially supported by the Public Welfare Industry Research Funds of China Meteorological Bureau (201206031).

ZHAOWenfang, born in 1980, M. S., senior engineer. Her research interests include big data, cloud computing, machine learning, meteorological big data processing.

LIUXulin, born in 1963, M. S., research fellow. His research interests include high performance computing, software architecture, data diming, knowledge discovery.