流式計算技術(shù)在山東省非考核地面氣象自動站數(shù)據(jù)實時處理中的應(yīng)用＊

馮勇 李微 袁雅涵

1.山東省氣象局氣象防災(zāi)減災(zāi)重點實驗室；2.山東省氣象信息中心；3.濟南高新技術(shù)產(chǎn)業(yè)開發(fā)區(qū)氣象局

本文基于山東省氣象大數(shù)據(jù)云平臺“天擎”，采用Storm流式處理技術(shù)，從數(shù)據(jù)收集、數(shù)據(jù)解碼、數(shù)據(jù)入庫、數(shù)據(jù)監(jiān)控等方面設(shè)計數(shù)據(jù)處理原型，解決2021年4月標(biāo)準(zhǔn)格式切換后非考核地面氣象自動站數(shù)據(jù)無法提供數(shù)據(jù)服務(wù)的難題，并與多線程處理程序進行時效對比，實踐表明，基于Storm的數(shù)據(jù)處理程序更能滿足現(xiàn)有業(yè)務(wù)系統(tǒng)對數(shù)據(jù)時效的要求。

氣象自動站觀測數(shù)據(jù)屬于氣象數(shù)據(jù)中的一種，具有產(chǎn)生快、實時性強、數(shù)據(jù)量大等特點[1]。經(jīng)過多年建設(shè)，山東省氣象部門已建成區(qū)域地面氣象自動站1921個，其中納入全省業(yè)務(wù)考核站點1763個，服務(wù)地方發(fā)展的非考核站點158個，覆蓋山東全省16地市，在城市防災(zāi)減災(zāi)、氣候變化、氣象服務(wù)中提供基礎(chǔ)數(shù)據(jù)支撐。

山東省氣象大數(shù)據(jù)云平臺“天擎”系統(tǒng)始建于2019年12月，作為全國試點部署省份，山東省氣象信息中心率先在全國完成業(yè)務(wù)部署，為全省提供“數(shù)算一體”的平臺化服務(wù)。云平臺數(shù)據(jù)全面，涵蓋CIMISS[2]所有數(shù)據(jù)，支撐氣象部門所有數(shù)據(jù)資源的匯聚、管理和服務(wù)，包括觀測數(shù)據(jù)、業(yè)務(wù)產(chǎn)品、政務(wù)信息、行業(yè)社會數(shù)據(jù)以及多媒體、出版物、設(shè)備信息、監(jiān)控日志等；實現(xiàn)云上計算能力，提供除數(shù)值預(yù)報以外各類數(shù)據(jù)產(chǎn)品加工、挖掘分析的平臺計算服務(wù)；直接支撐應(yīng)用，助力“云+端”業(yè)務(wù)模式的全面發(fā)展，構(gòu)成集約化、標(biāo)準(zhǔn)化、開放發(fā)展的氣象新業(yè)態(tài)，為山東省業(yè)務(wù)技術(shù)體制改革，建立“云+端”的業(yè)務(wù)模式新格局，奠定重要技術(shù)基礎(chǔ)[3]。

本文結(jié)合山東省氣象局的氣象預(yù)報、公共氣象服務(wù)等業(yè)務(wù)對非考核地面氣象自動站的需求，重點研究了Storm框架下的自動站實時處理技術(shù)，設(shè)計了非考核地面氣象自動站流式入庫算法，實現(xiàn)了一種分布式的地面自動氣象站數(shù)據(jù)流式處理，并基于山東省氣象大數(shù)據(jù)云平臺“天擎”，為全省氣象部門、行業(yè)用戶提供數(shù)據(jù)服務(wù)。

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

2021年以前，山東省氣象局主要依托新Z文件為全省業(yè)務(wù)系統(tǒng)提供非考核地面氣象自動站數(shù)據(jù)服務(wù)，2021年4月，山東省完成地面氣象自動站標(biāo)準(zhǔn)格式業(yè)務(wù)升級工作，原有TXT格式數(shù)據(jù)文件轉(zhuǎn)為BUFR文件，后續(xù)將取消新Z文件服務(wù)。目前，山東省氣象局主要依托氣象大數(shù)據(jù)云平臺（CMADaaS）實現(xiàn)全省1763個地面氣象自動站數(shù)據(jù)的實時處理、入庫存儲管理，并通過“天擎”的氣象數(shù)據(jù)統(tǒng)一服務(wù)接口（Meteorological Unified Service Interface Community，MUSIC）獲取實時數(shù)據(jù)，但是對于非考核地面氣象自動站BUFR格式數(shù)據(jù)，全省氣象業(yè)務(wù)系統(tǒng)對于BUFR格式數(shù)據(jù)文件缺少統(tǒng)一數(shù)據(jù)處理模塊，無法滿足全省氣象部門對非考核自動站數(shù)據(jù)的服務(wù)需求，迫切需要改進。

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

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

Storm是Twitter開源的分布式實時大數(shù)據(jù)處理框架，被業(yè)界稱為實時版Hadoop。隨著越來越多的場景對Hadoop的MapReduce高延遲無法容忍，比如網(wǎng)站統(tǒng)計、推薦系統(tǒng)、預(yù)警系統(tǒng)、金融系統(tǒng)(高頻交易、股票)等，大數(shù)據(jù)實時處理解決方案（流計算）的應(yīng)用日趨廣泛，目前已是分布式技術(shù)領(lǐng)域最新爆發(fā)點，而Storm更是流計算技術(shù)中的佼佼者和主流。主要由以下關(guān)鍵節(jié)點構(gòu)成。

（1）Nimbus（主節(jié)點）：主節(jié)點通常運行一個后臺程序——Nimbus，用于響應(yīng)分布在集群中的節(jié)點，分配任務(wù)和監(jiān)測故障。這個很類似于Hadoop中的JobTracker。

（2）Supervisor（工作節(jié)點）：工作節(jié)點同樣會運行一個后臺程序——Supervisor，用于收聽工作指派并基于要求運行工作進程。每個工作節(jié)點都是Topology中一個子集的實現(xiàn)。而Nimbus和Supervisor之間的協(xié)調(diào)則通過Zookeeper系統(tǒng)或者集群。

（3）Spout:從來源處讀取數(shù)據(jù)并放入Topology。Spout分成可靠和不可靠兩種；當(dāng)Storm接收失敗時，可靠的Spout會對Tuple（元組，數(shù)據(jù)項組成的列表）進行重發(fā)；而不可靠的Spout不會考慮接收成功與否只發(fā)射一次。

（4）Bolt：Topology中所有的處理都由Bolt完成。Bolt從Spout中接收數(shù)據(jù)并進行處理，如果遇到復(fù)雜流的處理也可能將Tuple發(fā)送給另一個Bolt進行處理。

2.2 邏輯表結(jié)構(gòu)設(shè)計

山東省非考核地面氣象自動站數(shù)據(jù)邏輯表總共分為3張：中國地面逐小時資料（山東省非考核）、中國地面分鐘資料（山東省非考核）、中國地面分鐘降水資料（山東省非考核），表結(jié)構(gòu)設(shè)計采用傳統(tǒng)的表字段模式，存儲結(jié)構(gòu)的命名主要參考行業(yè)標(biāo)準(zhǔn)《QX/T 233－2014氣象數(shù)據(jù)庫存儲管理命名》、《QX/T 133－2011氣象要素分類與編碼》等[4]。以中國地面分鐘降水資料（山東省非考核）為例，具體表結(jié)構(gòu)設(shè)計如表1所示。

表1 中國地面分鐘降水資料（山東省非考核）邏輯表結(jié)構(gòu)設(shè)計Tab.1 Logical table structure design for Chinese surface minute precipitation data (non-assessed in Shandong Province)

2.3 監(jiān)視信息DI接入設(shè)計

數(shù)據(jù)入庫監(jiān)視主要包括業(yè)務(wù)系統(tǒng)自身的產(chǎn)品生成、加工處理等運行狀況。DI信息采用POST方式發(fā)送至天鏡Restful接口。數(shù)據(jù)每次提交最多200條。格式如下：

2.4 系統(tǒng)結(jié)構(gòu)設(shè)計與實現(xiàn)

本文將Storm技術(shù)結(jié)合到氣象大數(shù)據(jù)云平臺“天擎”解碼入庫系統(tǒng)，利用Storm框架進行非考核地面自動站氣象數(shù)據(jù)實時處理，完成毫秒級數(shù)據(jù)入庫，并實時將監(jiān)控消息發(fā)送至氣象綜合業(yè)務(wù)實時監(jiān)控系統(tǒng)（天鏡）[5]。系統(tǒng)主要包括4部分：

（1）數(shù)據(jù)采集模塊：通過CTS將非考核小時數(shù)據(jù)、分鐘數(shù)據(jù)推送至氣象大數(shù)據(jù)云平臺“天擎”FTP節(jié)點，通過消息轉(zhuǎn)發(fā)與拆分系統(tǒng)分發(fā)至隊列SURF_PQC_A.0001.0041.R002_001和SURF_PQC_A.0001.0042.R002_001。

（2）數(shù)據(jù)解碼模塊：通過文件絕對路徑找尋文件進行數(shù)據(jù)解碼處理。

（3）數(shù)據(jù)入庫模塊：采用批量+定時的方式進行數(shù)據(jù)入庫處理，并將DI發(fā)送至天鏡。

（4）DIEI發(fā)送模塊：收集數(shù)據(jù)處理DI并通過Transfer程序發(fā)送至天鏡。

數(shù)據(jù)采集模塊復(fù)用數(shù)據(jù)收集與分發(fā)系統(tǒng)（CTS），省級區(qū)域中心站將非考核自動站BUFR數(shù)據(jù)打包后以文件形式發(fā)送至CTS數(shù)據(jù)入口Incoming目錄，通過文件名匹配+四級編碼，進行文件分發(fā)過濾，以FTP方式將打包文件轉(zhuǎn)發(fā)至氣象大數(shù)據(jù)云平臺數(shù)據(jù)入口/space/dpc/work/input，同時將MQ消息轉(zhuǎn)發(fā)至氣象大數(shù)據(jù)云平臺消息服務(wù)器。

數(shù)據(jù)解碼模塊、數(shù)據(jù)入庫模塊、DIEI發(fā)送模塊均基于Storm集群進行搭建，Storm集群部署在氣象大數(shù)據(jù)云平臺“天擎”DPC16-18節(jié)點。節(jié)點選用2U機架式X86服務(wù)器，CPU 32核，內(nèi)存256GB。其中主節(jié)點設(shè)置在服務(wù)器DPC16上，負責(zé)運行Nimbus程序，并分配數(shù)據(jù)處理任務(wù)給工作節(jié)點。工作節(jié)點設(shè)置在服務(wù)器DPC16-DPC18上，工作節(jié)點運行了狀態(tài)監(jiān)控程序(Supervisior程序)，負責(zé)監(jiān)聽Nimbus分配的任務(wù)[6]。當(dāng)來自主節(jié)點的數(shù)據(jù)解碼任務(wù)被提交時，Nimbus程序?qū)鶕?jù)工作量設(shè)定相應(yīng)的Task數(shù)量。

當(dāng)來自氣象大數(shù)據(jù)云平臺“天擎”的MQ消息通過主節(jié)點上的Spout程序傳遞至工作節(jié)點后，工作節(jié)點DPC16-DPC18上的運行Supervisior程序通過監(jiān)聽主節(jié)點的Spout程序，獲取山東省非考核地面氣象自動站打包文件中的Message信息，通過解析Message獲取CTS編碼、資料時間、文件名、文件絕對路徑等，然后傳輸給不同的Bolt程序進行處理[6]。如圖1所示，工作節(jié)點上的Bolt－解碼程序完成文件處理后，由Bolt--入庫程序?qū)⒔獯a程序生成的List體通過JDBC進行完成數(shù)據(jù)庫寫入；同時，Bolt-DIEI程序在整個數(shù)據(jù)解碼入庫過程中將發(fā)送的EIDI信息傳給“天鏡”Transfer接口，記錄數(shù)據(jù)處理過程全狀態(tài)。

圖1 基于Storm的山東省非考核地面自動站數(shù)據(jù)實時處理流程圖Fig.1 Storm-based real-time processing flow chart of non-assessment ground automatic station data in Shandong Province

在實際過程中，為了方便數(shù)據(jù)處理，區(qū)域站中心站對資料進行打包時分為分鐘數(shù)據(jù)和小時數(shù)據(jù)，因此在入庫的時候可根據(jù)數(shù)據(jù)庫表結(jié)構(gòu)的不同簡化入庫流程，根據(jù)資料類型來啟動對應(yīng)入庫程序。如圖2所示。

圖2 基于Storm的山東省非考核地面自動站數(shù)據(jù)實時處理流程圖Fig.2 Storm-based real-time processing flow chart of non-assessment ground automatic station data in Shandong

3 性能分析

3.1 省級入庫時效對比

以山東省非考核地面氣象自動站實時數(shù)據(jù)作為數(shù)據(jù)源，分別利用Storm解碼入庫和Java多線程入庫處理技術(shù)，對兩種不同方法的入庫時效進行對比。其中Storm集 群 配 置 了3個Work節(jié) 點，6個Spout和48個Blot；多線程處理程序部署在DPC14-DPC15上，機器配置：CPU16核×2、內(nèi)存256G。具體入庫時效對比結(jié)果如表2所示。可以看出，基于Storm的數(shù)據(jù)解碼處理程序時效均比多線程處理程序高4倍以上。

表2 Storm解碼入庫程序和多線程入庫程序時效對比Tab.2 Timing comparison between Storm decoding entry procedure and multi-threaded entry procedure

以山東省非考核地面氣象自動站歷史數(shù)據(jù)作為數(shù)據(jù)源，分別利用Storm解碼入庫和多線程入庫處理技術(shù)，對兩種不同方法的入庫效率進行對比。其中歷史小時BUFR數(shù)據(jù)15810條，歷史分鐘BUFR數(shù)據(jù)33922條。具體入庫效率對比結(jié)果如表3所示。可以看出，基于Storm的數(shù)據(jù)解碼處理程序入庫效率比多線程處理程序高6倍以上。

表3 Storm解碼入庫程序和多線程入庫程序入庫效率對比Tab.3 Comparison of accession efficiency between Storm decoding accession procedure and multi-threaded accession procedure

3.2 非功能性對比

在非功能性性能方面，Storm采用流式處理技術(shù)，實現(xiàn)了非考核站點數(shù)據(jù)1min到達預(yù)報員桌面的需求；在可靠性和穩(wěn)定性上要達到每條數(shù)據(jù)準(zhǔn)確入庫、記載錯誤、及時處理的流程在可拓展性上達到方便的應(yīng)對業(yè)務(wù)及數(shù)據(jù)庫的分布式拓展，可靈活調(diào)整入庫配置。在這些方面，Storm是更優(yōu)于多線程處理技術(shù)的。Storm 的實現(xiàn)方式及與多線程處理的實現(xiàn)方式如表4所示。

表4 Storm解碼程序和多線程處理程序的非功能性對比Tab.4 Non-functional comparison of Storm decoder and multi-threaded processor

4 結(jié)論

本文通過對Storm技術(shù)進行理論和業(yè)務(wù)邏輯設(shè)計，并與多線程入庫程序進行對比，事實證明，Storm在入庫時效、穩(wěn)定性、可靠性和可擴展性上都優(yōu)于多線程處理技術(shù)。Storm集群已在氣象大數(shù)據(jù)云平臺“天擎”穩(wěn)定運行379天，累計處理近3000萬份非結(jié)構(gòu)化地面氣象自動站數(shù)據(jù)，累計入庫條數(shù)近5500萬條，為全省氣象部門、行業(yè)用戶累計提供數(shù)據(jù)訪問近400萬次，有效支撐氣象防災(zāi)減災(zāi)、智慧城市建設(shè)，為全省開展氣象數(shù)據(jù)業(yè)務(wù)集約化奠定基礎(chǔ)。

引用

[1] 趙文芳,劉旭林.Spark Streaming框架下的氣象自動站數(shù)據(jù)實時處理系統(tǒng)[J].計算機應(yīng)用,2018,38(1):38-43+55.

[2] 趙芳,何文春,張小纓,等.全國綜合氣象信息共享平臺建設(shè)[J].氣象科技進展,2018,8(1):171-180.

[3] 馮勇,李微,朱輝,等.云計算環(huán)境下山東省氣象大數(shù)據(jù)云平臺的設(shè)計與實現(xiàn)[J].信息技術(shù)與信息化,2021(5):147-150.

[4] 徐擁軍,何文春,劉媛媛,等.氣象大數(shù)據(jù)存儲體系設(shè)計與實現(xiàn)[J].電子測量技術(shù),2020,43(22):19-25.

[5] 孫超,肖文名,陳永濤,等.氣象綜合業(yè)務(wù)實時監(jiān)控系統(tǒng)的設(shè)計[J].氣象科技進展,2018,8(1):153-157.

[6] 廖婷婷,王彪,肖衛(wèi)青,等.Storm流式技術(shù)在地面氣象數(shù)據(jù)處理中的應(yīng)用[J].中低緯山地氣象,2019,43(5):78-81.