基于消息的氣象服務數(shù)據(jù)加工流水線設計與應用

宋瑛瑛 李雁鵬 陳正廷 凌柏 馮宇星 曹宇釗

（中國氣象局公共氣象服務中心，北京 100081）

基于消息的氣象服務數(shù)據(jù)加工流水線設計與應用

宋瑛瑛 李雁鵬 陳正廷 凌柏 馮宇星 曹宇釗

（中國氣象局公共氣象服務中心，北京 100081）

梳理了公共氣象服務業(yè)務的數(shù)據(jù)流程，提出了一種基于消息的氣象服務數(shù)據(jù)加工流水線技術(shù)，用于提高氣象數(shù)據(jù)服務的時效性與數(shù)據(jù)資源更新的一致性。著重介紹了消息中間件、流水線架構(gòu)設計、交互文件標準等技術(shù)問題。結(jié)合分鐘級降水預報數(shù)據(jù)實例，初步展示了消息驅(qū)動數(shù)據(jù)流應用在氣象服務數(shù)據(jù)支撐系統(tǒng)中的優(yōu)勢。

氣象數(shù)據(jù)，消息中間件，交互文件標準，系統(tǒng)設計

0 引言

氣象服務是人類生產(chǎn)生活的基礎(chǔ)條件，它不僅與公眾生活密切相關(guān)，同時也直接影響政府決策、國民生產(chǎn)以及社會發(fā)展。隨著信息網(wǎng)絡技術(shù)的不斷發(fā)展，各行各業(yè)對氣象服務的要求越來越趨于精細化，隨之而來，分門別類的氣象服務產(chǎn)品層出不窮。為滿足各類應用需求，數(shù)據(jù)信息交互單元也逐漸變得龐雜。當前，在中國氣象局信息化建設的大背景下，公共氣象服務信息化建設的需求越來越緊迫，集中表現(xiàn)在從分散向集約化發(fā)展，提供統(tǒng)一的氣象數(shù)據(jù)服務。因此，本文梳理了國家級公共氣象服務業(yè)務的數(shù)據(jù)流，分析了現(xiàn)存問題，提出了基于消息的氣象服務數(shù)據(jù)加工流水線技術(shù)，旨在提高氣象數(shù)據(jù)流的時效性和魯棒性，從而更好地支撐氣象服務業(yè)務的快速發(fā)展。

1 消息中間件

隨著氣象業(yè)務的跨越式發(fā)展，當前中國氣象局公共氣象服務中心應用的氣象服務數(shù)據(jù)系統(tǒng)存在數(shù)據(jù)處理流程交叉、運行效率低、業(yè)務耦合性強等問題。本文提出利用消息驅(qū)動業(yè)務流程，解決上述問題。具體而言，利用消息中間件高效可靠的傳遞機制進行平臺無關(guān)的數(shù)據(jù)交換，基于數(shù)據(jù)通信進行分布式系統(tǒng)的集成。發(fā)送者將消息發(fā)送到消息服務器，消息服務器將消息存放在若干隊列中，在滿足觸發(fā)條件后再將消息轉(zhuǎn)發(fā)給接收者。消息傳輸模型分為點對點模型（PTP）和發(fā)布/訂閱模型（pub/sub），發(fā)布/訂閱模型具有異步、松耦合、多對多通信等特點。發(fā)布者廣播發(fā)送消息給中間代理，訂閱者只需去代理中接收自己感興趣的消息，發(fā)布者并不知道究竟是哪個訂閱者接收到了自己發(fā)布的消息。

目前，市面上常見的發(fā)布/訂閱模型消息中間件有：Kafka、ActiveMQ、OpenJMS和RabbitMQ等。國內(nèi)各氣象部門在應用消息中間件傳遞數(shù)據(jù)方面做了很多研究，例如：應用ActiveMQ技術(shù)縮短了預警信號發(fā)布時耗[1]，基于RabbitMQ消息中間件和元數(shù)據(jù)技術(shù)實現(xiàn)了多種氣象觀測資料的采集、傳輸、入庫和備份的統(tǒng)一處理[2]。通過業(yè)務調(diào)研，本文以Kafka消息中間件為核心，在集約資源的基礎(chǔ)上，兼顧性能，設計了基于消息的氣象服務數(shù)據(jù)加工流水線，用以提高數(shù)據(jù)流動效率，實現(xiàn)數(shù)據(jù)內(nèi)容統(tǒng)一。

2 氣象數(shù)據(jù)流

氣象數(shù)據(jù)流具有多來源、高并發(fā)及更新快等特點。針對這些特性，以往的氣象數(shù)據(jù)處理過程，多采用多進程并發(fā)技術(shù)[3]，基于web服務[4]，實現(xiàn)氣象數(shù)據(jù)實時共享；近年來，多采用消息驅(qū)動機制處理氣象數(shù)據(jù)[1-2,5]，以期提升傳輸效率。

2.1 數(shù)據(jù)流分析

公共氣象服務中心的氣象數(shù)據(jù)服務主要包括專業(yè)氣象和公眾氣象兩大部分。其中，專業(yè)氣象為預警信息發(fā)布、自然災害決策等提供數(shù)據(jù)服務；而公眾氣象為常規(guī)天氣預報制作提供數(shù)據(jù)支撐，主要服務對象有中國氣象頻道、中國天氣網(wǎng)、中國天氣通APP等媒體和天氣服務軟件。在現(xiàn)有氣象數(shù)據(jù)服務系統(tǒng)中，存在以下不足之處：1）兩大業(yè)務存在數(shù)據(jù)源重疊交叉，數(shù)據(jù)重復采集、多次加工，數(shù)據(jù)樣本多重復制，在極端條件下，會出現(xiàn)氣象服務產(chǎn)品數(shù)據(jù)內(nèi)容不一致的情況。2）關(guān)聯(lián)任務執(zhí)行時存在任務堆積、耗時長等問題。在部分業(yè)務執(zhí)行時間不確定的條件下，為了保障業(yè)務數(shù)據(jù)的完整性，需延長定制計劃任務時間范圍，這就增加了后續(xù)處理環(huán)節(jié)的等待時間，造成延遲。3）對超出定制時間范圍的數(shù)據(jù)無法實現(xiàn)自動處理，只能通過手動執(zhí)行應急處理，無法做到全自動無人值守。4）數(shù)據(jù)讀寫操作頻繁，增加服務器的并發(fā)負載，造成資源浪費。可見，定時計劃任務是拖延數(shù)據(jù)流動的重要一環(huán)，而消息即來即走、分布式并發(fā)等特性正可以解決這些問題。

2.2 消息驅(qū)動數(shù)據(jù)流水線

基于消息驅(qū)動的數(shù)據(jù)流，將流動過程分解成主題，通過發(fā)布/訂閱模式，即時發(fā)布主題至消息總線，計算集群監(jiān)聽消息，即時處理主題，縮短了數(shù)據(jù)流動時間。這種分布式、高并發(fā)、無延遲的集約型數(shù)據(jù)流為下游氣象產(chǎn)品制作業(yè)務提供即時的數(shù)據(jù)支撐。如圖1所示。

1）業(yè)務數(shù)據(jù)流解耦。根據(jù)業(yè)務邏輯關(guān)系，將數(shù)據(jù)流分解為多個獨立執(zhí)行單元，即消息。同時為每個消息建立主題（topic），例如數(shù)據(jù)讀取主題、數(shù)據(jù)處理主題、數(shù)據(jù)發(fā)布主題等。消息主題不具有業(yè)務意義，僅作為消息總線上的標識使用。

2）數(shù)據(jù)即時加工處理。生產(chǎn)者（producer）發(fā)布主題至消息總線，消費者（consumer）隨時監(jiān)聽消息總線上已訂閱的主題，一旦發(fā)現(xiàn)觸發(fā)消息，立即執(zhí)行。本文中的消費者由多臺物理機或虛擬機組成基于SSH協(xié)議的計算集群實現(xiàn)，集群中的每個節(jié)點是無狀態(tài)的，可并行消費多個主題。

3）數(shù)據(jù)流分發(fā)。處理完成的數(shù)據(jù)按需分發(fā)至下游業(yè)務系統(tǒng)。

圖1 數(shù)據(jù)流水線Fig. 1 Data flow

2.3 兩種數(shù)據(jù)流程特性比較

消息驅(qū)動數(shù)據(jù)流摒棄了定時計劃任務模式，采用了發(fā)布/訂閱消息模式，不斷地從消息總線上監(jiān)聽消息，一旦接收到消息即刻觸發(fā)執(zhí)行主題任務。由計算集群采用分布式計算完成任務的并發(fā)操作，改變了傳統(tǒng)數(shù)據(jù)流集中控制、順序執(zhí)行任務的處理過程，極大地提高了數(shù)據(jù)處理效率。目前，氣象消息總線的主要應用是業(yè)務集成和數(shù)據(jù)共享。從業(yè)務集成方面看，可以有效解耦氣象數(shù)據(jù)流處理過程，提高信息交互能力和功能復用水平。從數(shù)據(jù)共享方面看，消息總線為數(shù)據(jù)的及時更新和一致性提供了可靠保障。兩種數(shù)據(jù)流程特性比較如表1所示。

表1 兩種數(shù)據(jù)流特性對比Table1 Comparisons between two kinds of data flow

3 系統(tǒng)設計與應用

3.1 系統(tǒng)架構(gòu)

為了能更好地集成現(xiàn)有業(yè)務數(shù)據(jù)流處理過程，基于消息的氣象服務數(shù)據(jù)加工流水線設計采用三層架構(gòu)實現(xiàn)，分為數(shù)據(jù)層、消息中間層和業(yè)務層（圖2）。

圖2 系統(tǒng)架構(gòu)Fig. 2 System architecture

數(shù)據(jù)層是氣象服務數(shù)據(jù)加工流水線的基礎(chǔ)。數(shù)據(jù)文件存儲在大型分布式物理設備中，計算集群是數(shù)據(jù)流各環(huán)節(jié)的真正執(zhí)行者，將存儲和計算隔離開，實現(xiàn)了數(shù)據(jù)與計算的解耦。

消息中間層在本架構(gòu)中具有承上啟下的作用，在Kafka消息總線的基礎(chǔ)上實現(xiàn)業(yè)務運行框架。將一條業(yè)務數(shù)據(jù)流處理過程分解為啟動和執(zhí)行兩部分，啟動模塊是處理過程的起點，是消息總線上的第一條消息，執(zhí)行模塊監(jiān)聽到該消息后分步驟并行處理業(yè)務程序代碼，將結(jié)果消息發(fā)送至總線。本文將啟動和執(zhí)行模塊封裝為消息觸發(fā)器和業(yè)務執(zhí)行器。

根據(jù)應用需求，本文設計了兩種消息觸發(fā)器：1）采集數(shù)據(jù)：根據(jù)業(yè)務配置開始時間和間隔，輪詢向上游數(shù)據(jù)提出采集要求，采集完成后即刻發(fā)出指令，啟動本次業(yè)務數(shù)據(jù)流處理過程；2）監(jiān)控數(shù)據(jù)：根據(jù)業(yè)務配置目錄和文件（可多個），監(jiān)控數(shù)據(jù)狀態(tài)，完整到達后即刻向消息總線發(fā)出消息，觸發(fā)執(zhí)行下一環(huán)節(jié)消息主題。

業(yè)務執(zhí)行器由消息收發(fā)、消息過濾和業(yè)務程序執(zhí)行代碼組成：1）消息收發(fā)：計算集群各節(jié)點監(jiān)聽訂閱的主題消息，處理完成后，向消息總線發(fā)送本環(huán)節(jié)消息結(jié)果；2）消息過濾：過濾重復消息，防止反復觸發(fā)造成消息異常堆積；3）業(yè)務程序執(zhí)行代碼：分解數(shù)據(jù)流處理過程，封裝業(yè)務單元程序執(zhí)行代碼，由計算集群并行處理。

服務層是氣象服務數(shù)據(jù)加工流水線的窗口。配置服務為運行框架提供集群、業(yè)務和消息等參數(shù)配置；日志服務記錄業(yè)務運行狀態(tài)；監(jiān)控服務為運維提供了可視化用戶界面。

3.2 關(guān)鍵技術(shù)

為解決數(shù)據(jù)處理時效性問題，本文設計了一種基于Kafka消息總線的業(yè)務運行框架，實現(xiàn)了氣象服務數(shù)據(jù)即時處理；為解決業(yè)務層與運行框架間的數(shù)據(jù)交互問題，制定了規(guī)范性xml消息配置，提供了統(tǒng)一的參數(shù)標準。

3.2.1 基于Kafka的消息驅(qū)動

Kafka包括生產(chǎn)者、消費者和代理者三個部分。生產(chǎn)者生產(chǎn)的消息被放在主題中，一個主題可以設置多個分區(qū)，推送消息給代理者，等待消費者接收；消費者在想要訂閱消息時，向代理者發(fā)出請求，告知其主題和分區(qū)值，主動拉取消息（圖3）。數(shù)據(jù)流經(jīng)過解耦，運行在消息總線上，經(jīng)多個計算集群即時執(zhí)行處理，迅速將加工數(shù)據(jù)分發(fā)出去。

圖3 消息發(fā)布/訂閱簡圖Fig. 3 Message publish/subscribe

以Java代碼為例，介紹消息的生產(chǎn)者和消費者。

1） 創(chuàng)建生產(chǎn)者

// 創(chuàng)建生產(chǎn)者屬性參數(shù)

Properties dps = new Properties();

//指定代理服務器

dps.put("metadata.broker.list",kafkahosts.toString()); //用該屬性參數(shù)創(chuàng)建生產(chǎn)者

Producer〈String, String〉 producer = new Producer〈String, String〉(new ProducerConf i g(MsgController.dps));

//生產(chǎn)者廣播發(fā)送topic/link消息至消息總線

producer.send(new KeyedMessage〈String, String〉(topic, new Date().getTime() + "" ,next-link-id));

2）創(chuàng)建消費者

// 創(chuàng)建消費者屬性參數(shù)

Properties cps = new Properties();

//指定消費組

cps.put("group.id", group);

//消費者zookeeper 配置

cps.put("zookeeper.connect", zookeeperhosts. toString());

//用消費者屬性參數(shù)創(chuàng)建消費者

ConsumerConnector consumer = kafka.consumer. Consumer.createJavaConsumerConnector(new ConsumerConf i g(cps));

3.2.2 消息配置格式

xml文件提供統(tǒng)一的方法來描述結(jié)構(gòu)化數(shù)據(jù)，通常作為異構(gòu)系統(tǒng)間數(shù)據(jù)交換的格式[6]。參數(shù)配置采用xml文件描述，一個業(yè)務對應配置一個xml文件，設計xml文件命名規(guī)則為業(yè)務種類（2位數(shù)字）+業(yè)務序列（5位數(shù)字）.xml，例如：1010001.xml。

消息配置內(nèi)容由三部分組成：

1）service：業(yè)務基本信息。name：業(yè)務名稱；type分兩類：采集（collection）和監(jiān)控（notify），與消息觸發(fā)器類型對應。

2）type描述：分為采集和監(jiān)控標簽，為消息觸發(fā)器提供所需參數(shù)信息。①采集標簽：description：描述采集業(yè)務詳情；start-time：開始時間；period：時間間隔。②監(jiān)控標簽：description：描述監(jiān)控業(yè)務詳情；path：要監(jiān)控的文件目錄；f i les：監(jiān)控的數(shù)據(jù)源文件。

3）link：主題參數(shù)。id：標識主題單元號，由業(yè)務種類（2位數(shù)字）+業(yè)務序列（5位數(shù)字）+環(huán)節(jié)號（3位數(shù)字）共10位數(shù)字組成；description：描述該環(huán)節(jié)主要操作內(nèi)容；order：描述主題執(zhí)行命令代碼；next-link：指定下一topic/link對。

3.3 業(yè)務應用

實時天氣預報是網(wǎng)站和移動終端用戶的重點選擇氣象服務，具有廣泛的用戶群體和極大的需求。本節(jié)以雷達分鐘級降水系統(tǒng)（PM-MQPF）數(shù)據(jù)為例，闡述業(yè)務數(shù)據(jù)流實際應用。分鐘級降水數(shù)據(jù)來源于214部雷達站逐6分鐘實時觀測產(chǎn)生的基數(shù)據(jù)，經(jīng)過計算處理，形成臨近降水預報服務產(chǎn)品。在現(xiàn)有讀取基數(shù)據(jù)、前處理、主處理、拼圖4個處理程序基礎(chǔ)上，通過Kafka消息驅(qū)動機制進一步將這四部分分解剝離，形成獨立單元流水線作業(yè)模式，通過單元返回值觸發(fā)后續(xù)單元主題。將在整個數(shù)據(jù)計算加工過程中耗時較長的前處理部分，運用計算集群并行處理，以提高計算效率。

1）分解PM-MQPF數(shù)據(jù)處理過程。封裝為基數(shù)據(jù)、前處理、主處理和拼圖四個部分，為每部分增加返回值。

2）配置PM-MQPF數(shù)據(jù)消息參數(shù)。為監(jiān)控指定目錄和文件，為業(yè)務環(huán)節(jié)分配主題，T1（前處理），T2（主處理），T3（拼圖）。

3）消息總線監(jiān)聽文件，有文件到達即發(fā)送消息。

4）計算集群監(jiān)聽消息，即時處理主題。T1分布式并行對每個數(shù)據(jù)文件進行前處理；T2統(tǒng)計每6分鐘前處理完成的情況，處理量完成80%以后發(fā)送消息啟動主處理，主處理完成之后再發(fā)送消息；觸發(fā)T3加工生成拼圖，完成后通知下游環(huán)節(jié)進行產(chǎn)品分發(fā)。

PM-MQPF分鐘級降水數(shù)據(jù)流水線采用消息機制和計算集群，分解封裝處理單元，降低了各處單元的耦合度，提高了系統(tǒng)運行效率。因此，氣象信息總線的應用不僅降低了系統(tǒng)開發(fā)和運行成本，也提高了業(yè)務應用的實時性，同時為后續(xù)拓展應用服務提供了空間。

4 結(jié)束語

基于消息的氣象服務數(shù)據(jù)加工流水線降低了業(yè)務耦合度，優(yōu)化了氣象數(shù)據(jù)流程，為進一步實現(xiàn)氣象數(shù)據(jù)實時同步、提高數(shù)據(jù)流時效性奠定了必要基礎(chǔ)。當前，系統(tǒng)在功能、性能和穩(wěn)定性方面都有優(yōu)異的表現(xiàn)，不但能夠滿足下游各氣象業(yè)務的實際需求，同時為業(yè)務拓展提供了空間。在后續(xù)的研究計劃中，計算集群的高并發(fā)、負載均衡等問題將被進一步探索。

[1]錢崢, 胡亞旦, 黃旋旋. 基于“消息中間件”技術(shù)的氣象信息總線. 氣象科技, 2016, 44(2): 217-222.

[2]韓笑, 王力, 王吉濱, 等. 一種地市級氣象數(shù)據(jù)庫的設計與應用. 氣象科技, 2015, 43(6): 1053-1059.

[3]胡英楣, 沈文海, 宋之光. 多進程并發(fā)在國內(nèi)氣象通信系統(tǒng)的應用. 應用氣象學報, 2007, 18(6): 877-884.

[4]王甫棣, 林潤生, 胡英楣. 基于Web服務的氣象數(shù)據(jù)服務. 計算機工程, 2009, 35(8): 280-282.

[5]王力, 韓笑, 劉培寧, 等. 基于MQ的氣象數(shù)據(jù)采集與監(jiān)控系統(tǒng)設計. 氣象科技, 2015, 43(3): 451-457.

[6]曹衛(wèi). XML技術(shù)在氣象信息發(fā)布系統(tǒng)中的應用. 福建電腦, 2009, 25(4): 110-111.

Design and Application of Message-Driven Data Processing Flow in Meteorological Service

Song Yingying, Li Yanpeng, Chen Zhengting, Ling Bai, Feng Yuxing, Cao Yuzhao
(Public Meteorological Service Center of China Meteorological Administration, Beijing 100081)

Based on the analysis of public meteorological service data fl ow, this paper proposes a message-driven data processing technology in the meteorological service to improve the timeliness of meteorological data services and the uniformity of data resources. The key technologies, such as message middleware, system architecture, interactive fi le standard, are introduced here. An example of PM-MQPF data is described in detail to express the advantages of data fl ow, which was driven by messages in meteorological data services system.

meteorological data, message-oriented middleware, interactive fi le standard, system design

10.3969/j.issn.2095-1973.2016.06.008

2016年9月21日；

2016年11月25日

宋瑛瑛（1983—），Email：syy0822@126.com

資助信息：中國氣象局公共氣象服務中心業(yè)務基金項目（K2016006）