流式计算技术在山东省非考核地面气象自动站数据实时处理中的应用＊

冯勇 李微 袁雅涵

1.山东省气象局气象防灾减灾重点实验室；2.山东省气象信息中心；3.济南高新技术产业开发区气象局

本文基于山东省气象大数据云平台“天擎”，采用Storm流式处理技术，从数据收集、数据解码、数据入库、数据监控等方面设计数据处理原型，解决2021年4月标准格式切换后非考核地面气象自动站数据无法提供数据服务的难题，并与多线程处理程序进行时效对比，实践表明，基于Storm的数据处理程序更能满足现有业务系统对数据时效的要求。

气象自动站观测数据属于气象数据中的一种，具有产生快、实时性强、数据量大等特点[1]。经过多年建设，山东省气象部门已建成区域地面气象自动站1921个，其中纳入全省业务考核站点1763个，服务地方发展的非考核站点158个，覆盖山东全省16地市，在城市防灾减灾、气候变化、气象服务中提供基础数据支撑。

山东省气象大数据云平台“天擎”系统始建于2019年12月，作为全国试点部署省份，山东省气象信息中心率先在全国完成业务部署，为全省提供“数算一体”的平台化服务。云平台数据全面，涵盖CIMISS[2]所有数据，支撑气象部门所有数据资源的汇聚、管理和服务，包括观测数据、业务产品、政务信息、行业社会数据以及多媒体、出版物、设备信息、监控日志等；实现云上计算能力，提供除数值预报以外各类数据产品加工、挖掘分析的平台计算服务；直接支撑应用，助力“云+端”业务模式的全面发展，构成集约化、标准化、开放发展的气象新业态，为山东省业务技术体制改革，建立“云+端”的业务模式新格局，奠定重要技术基础[3]。

本文结合山东省气象局的气象预报、公共气象服务等业务对非考核地面气象自动站的需求，重点研究了Storm框架下的自动站实时处理技术，设计了非考核地面气象自动站流式入库算法，实现了一种分布式的地面自动气象站数据流式处理，并基于山东省气象大数据云平台“天擎”，为全省气象部门、行业用户提供数据服务。

1 业务现状

2021年以前，山东省气象局主要依托新Z文件为全省业务系统提供非考核地面气象自动站数据服务，2021年4月，山东省完成地面气象自动站标准格式业务升级工作，原有TXT格式数据文件转为BUFR文件，后续将取消新Z文件服务。目前，山东省气象局主要依托气象大数据云平台（CMADaaS）实现全省1763个地面气象自动站数据的实时处理、入库存储管理，并通过“天擎”的气象数据统一服务接口（Meteorological Unified Service Interface Community，MUSIC）获取实时数据，但是对于非考核地面气象自动站BUFR格式数据，全省气象业务系统对于BUFR格式数据文件缺少统一数据处理模块，无法满足全省气象部门对非考核自动站数据的服务需求，迫切需要改进。

2 系统设计

2.1 系统技术框架

Storm是Twitter开源的分布式实时大数据处理框架，被业界称为实时版Hadoop。随着越来越多的场景对Hadoop的MapReduce高延迟无法容忍，比如网站统计、推荐系统、预警系统、金融系统(高频交易、股票)等，大数据实时处理解决方案（流计算）的应用日趋广泛，目前已是分布式技术领域最新爆发点，而Storm更是流计算技术中的佼佼者和主流。主要由以下关键节点构成。

（1）Nimbus（主节点）：主节点通常运行一个后台程序——Nimbus，用于响应分布在集群中的节点，分配任务和监测故障。这个很类似于Hadoop中的JobTracker。

（2）Supervisor（工作节点）：工作节点同样会运行一个后台程序——Supervisor，用于收听工作指派并基于要求运行工作进程。每个工作节点都是Topology中一个子集的实现。而Nimbus和Supervisor之间的协调则通过Zookeeper系统或者集群。

（3）Spout:从来源处读取数据并放入Topology。Spout分成可靠和不可靠两种；当Storm接收失败时，可靠的Spout会对Tuple（元组，数据项组成的列表）进行重发；而不可靠的Spout不会考虑接收成功与否只发射一次。

（4）Bolt：Topology中所有的处理都由Bolt完成。Bolt从Spout中接收数据并进行处理，如果遇到复杂流的处理也可能将Tuple发送给另一个Bolt进行处理。

2.2 逻辑表结构设计

山东省非考核地面气象自动站数据逻辑表总共分为3张：中国地面逐小时资料（山东省非考核）、中国地面分钟资料（山东省非考核）、中国地面分钟降水资料（山东省非考核），表结构设计采用传统的表字段模式，存储结构的命名主要参考行业标准《QX/T 233－2014气象数据库存储管理命名》、《QX/T 133－2011气象要素分类与编码》等[4]。以中国地面分钟降水资料（山东省非考核）为例，具体表结构设计如表1所示。

表1 中国地面分钟降水资料（山东省非考核）逻辑表结构设计Tab.1 Logical table structure design for Chinese surface minute precipitation data (non-assessed in Shandong Province)

2.3 监视信息DI接入设计

数据入库监视主要包括业务系统自身的产品生成、加工处理等运行状况。DI信息采用POST方式发送至天镜Restful接口。数据每次提交最多200条。格式如下：

2.4 系统结构设计与实现

本文将Storm技术结合到气象大数据云平台“天擎”解码入库系统，利用Storm框架进行非考核地面自动站气象数据实时处理，完成毫秒级数据入库，并实时将监控消息发送至气象综合业务实时监控系统（天镜）[5]。系统主要包括4部分：

（1）数据采集模块：通过CTS将非考核小时数据、分钟数据推送至气象大数据云平台“天擎”FTP节点，通过消息转发与拆分系统分发至队列SURF_PQC_A.0001.0041.R002_001和SURF_PQC_A.0001.0042.R002_001。

（2）数据解码模块：通过文件绝对路径找寻文件进行数据解码处理。

（3）数据入库模块：采用批量+定时的方式进行数据入库处理，并将DI发送至天镜。

（4）DIEI发送模块：收集数据处理DI并通过Transfer程序发送至天镜。

数据采集模块复用数据收集与分发系统（CTS），省级区域中心站将非考核自动站BUFR数据打包后以文件形式发送至CTS数据入口Incoming目录，通过文件名匹配+四级编码，进行文件分发过滤，以FTP方式将打包文件转发至气象大数据云平台数据入口/space/dpc/work/input，同时将MQ消息转发至气象大数据云平台消息服务器。

数据解码模块、数据入库模块、DIEI发送模块均基于Storm集群进行搭建，Storm集群部署在气象大数据云平台“天擎”DPC16-18节点。节点选用2U机架式X86服务器，CPU 32核，内存256GB。其中主节点设置在服务器DPC16上，负责运行Nimbus程序，并分配数据处理任务给工作节点。工作节点设置在服务器DPC16-DPC18上，工作节点运行了状态监控程序(Supervisior程序)，负责监听Nimbus分配的任务[6]。当来自主节点的数据解码任务被提交时，Nimbus程序将会根据工作量设定相应的Task数量。

当来自气象大数据云平台“天擎”的MQ消息通过主节点上的Spout程序传递至工作节点后，工作节点DPC16-DPC18上的运行Supervisior程序通过监听主节点的Spout程序，获取山东省非考核地面气象自动站打包文件中的Message信息，通过解析Message获取CTS编码、资料时间、文件名、文件绝对路径等，然后传输给不同的Bolt程序进行处理[6]。如图1所示，工作节点上的Bolt－解码程序完成文件处理后，由Bolt--入库程序将解码程序生成的List体通过JDBC进行完成数据库写入；同时，Bolt-DIEI程序在整个数据解码入库过程中将发送的EIDI信息传给“天镜”Transfer接口，记录数据处理过程全状态。

图1 基于Storm的山东省非考核地面自动站数据实时处理流程图Fig.1 Storm-based real-time processing flow chart of non-assessment ground automatic station data in Shandong Province

在实际过程中，为了方便数据处理，区域站中心站对资料进行打包时分为分钟数据和小时数据，因此在入库的时候可根据数据库表结构的不同简化入库流程，根据资料类型来启动对应入库程序。如图2所示。

图2 基于Storm的山东省非考核地面自动站数据实时处理流程图Fig.2 Storm-based real-time processing flow chart of non-assessment ground automatic station data in Shandong

3 性能分析

3.1 省级入库时效对比

以山东省非考核地面气象自动站实时数据作为数据源，分别利用Storm解码入库和Java多线程入库处理技术，对两种不同方法的入库时效进行对比。其中Storm集 群 配 置 了3个Work节 点，6个Spout和48个Blot；多线程处理程序部署在DPC14-DPC15上，机器配置：CPU16核×2、内存256G。具体入库时效对比结果如表2所示。可以看出，基于Storm的数据解码处理程序时效均比多线程处理程序高4倍以上。

表2 Storm解码入库程序和多线程入库程序时效对比Tab.2 Timing comparison between Storm decoding entry procedure and multi-threaded entry procedure

以山东省非考核地面气象自动站历史数据作为数据源，分别利用Storm解码入库和多线程入库处理技术，对两种不同方法的入库效率进行对比。其中历史小时BUFR数据15810条，历史分钟BUFR数据33922条。具体入库效率对比结果如表3所示。可以看出，基于Storm的数据解码处理程序入库效率比多线程处理程序高6倍以上。

表3 Storm解码入库程序和多线程入库程序入库效率对比Tab.3 Comparison of accession efficiency between Storm decoding accession procedure and multi-threaded accession procedure

3.2 非功能性对比

在非功能性性能方面，Storm采用流式处理技术，实现了非考核站点数据1min到达预报员桌面的需求；在可靠性和稳定性上要达到每条数据准确入库、记载错误、及时处理的流程在可拓展性上达到方便的应对业务及数据库的分布式拓展，可灵活调整入库配置。在这些方面，Storm是更优于多线程处理技术的。Storm 的实现方式及与多线程处理的实现方式如表4所示。

表4 Storm解码程序和多线程处理程序的非功能性对比Tab.4 Non-functional comparison of Storm decoder and multi-threaded processor

4 结论

本文通过对Storm技术进行理论和业务逻辑设计，并与多线程入库程序进行对比，事实证明，Storm在入库时效、稳定性、可靠性和可扩展性上都优于多线程处理技术。Storm集群已在气象大数据云平台“天擎”稳定运行379天，累计处理近3000万份非结构化地面气象自动站数据，累计入库条数近5500万条，为全省气象部门、行业用户累计提供数据访问近400万次，有效支撑气象防灾减灾、智慧城市建设，为全省开展气象数据业务集约化奠定基础。

引用

[1] 赵文芳,刘旭林.Spark Streaming框架下的气象自动站数据实时处理系统[J].计算机应用,2018,38(1):38-43+55.

[2] 赵芳,何文春,张小缨,等.全国综合气象信息共享平台建设[J].气象科技进展,2018,8(1):171-180.

[3] 冯勇,李微,朱辉,等.云计算环境下山东省气象大数据云平台的设计与实现[J].信息技术与信息化,2021(5):147-150.

[4] 徐拥军,何文春,刘媛媛,等.气象大数据存储体系设计与实现[J].电子测量技术,2020,43(22):19-25.

[5] 孙超,肖文名,陈永涛,等.气象综合业务实时监控系统的设计[J].气象科技进展,2018,8(1):153-157.

[6] 廖婷婷,王彪,肖卫青,等.Storm流式技术在地面气象数据处理中的应用[J].中低纬山地气象,2019,43(5):78-81.