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

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

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

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

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

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

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

氣象數據，消息中間件，交互文件標準，系統設計

0 引言

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

1 消息中間件

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

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

2 氣象數據流

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

2.1 數據流分析

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

2.2 消息驅動數據流水線

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

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

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

3）數據流分發。處理完成的數據按需分發至下游業務系統。

圖1 數據流水線Fig. 1 Data flow

2.3 兩種數據流程特性比較

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

表1 兩種數據流特性對比Table1 Comparisons between two kinds of data flow

3 系統設計與應用

3.1 系統架構

為了能更好地集成現有業務數據流處理過程，基于消息的氣象服務數據加工流水線設計采用三層架構實現，分為數據層、消息中間層和業務層（圖2）。

圖2 系統架構Fig. 2 System architecture

數據層是氣象服務數據加工流水線的基礎。數據文件存儲在大型分布式物理設備中，計算集群是數據流各環節的真正執行者，將存儲和計算隔離開，實現了數據與計算的解耦。

消息中間層在本架構中具有承上啟下的作用，在Kafka消息總線的基礎上實現業務運行框架。將一條業務數據流處理過程分解為啟動和執行兩部分，啟動模塊是處理過程的起點，是消息總線上的第一條消息，執行模塊監聽到該消息后分步驟并行處理業務程序代碼，將結果消息發送至總線。本文將啟動和執行模塊封裝為消息觸發器和業務執行器。

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

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

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

3.2 關鍵技術

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

3.2.1 基于Kafka的消息驅動

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

圖3 消息發布/訂閱簡圖Fig. 3 Message publish/subscribe

以Java代碼為例，介紹消息的生產者和消費者。

1） 創建生產者

// 創建生產者屬性參數

Properties dps = new Properties();

//指定代理服務器

dps.put("metadata.broker.list",kafkahosts.toString()); //用該屬性參數創建生產者

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

//生產者廣播發送topic/link消息至消息總線

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

2）創建消費者

// 創建消費者屬性參數

Properties cps = new Properties();

//指定消費組

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

//消費者zookeeper 配置

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

//用消費者屬性參數創建消費者

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

3.2.2 消息配置格式

xml文件提供統一的方法來描述結構化數據，通常作為異構系統間數據交換的格式[6]。參數配置采用xml文件描述，一個業務對應配置一個xml文件，設計xml文件命名規則為業務種類（2位數字）+業務序列（5位數字）.xml，例如：1010001.xml。

消息配置內容由三部分組成：

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

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

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

3.3 業務應用

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

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

2）配置PM-MQPF數據消息參數。為監控指定目錄和文件，為業務環節分配主題，T1（前處理），T2（主處理），T3（拼圖）。

3）消息總線監聽文件，有文件到達即發送消息。

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

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

4 結束語

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

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

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

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

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

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

[6]曹衛. XML技術在氣象信息發布系統中的應用. 福建電腦, 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

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