气象大数据分布式计算落地实践：从数据存储到性能调优

如果你跟我一样这几年一直在跟气象数据和分布式计算打交道你大概也会有同样的体会气象数据这口饭是真的越来越难吃了。遥感、雷达、自动站、数值模式……每类数据都是“量大、实时、格式怪”的典型代表单机方案早就扛不住分布式计算已经从“可选增强”变成了“必上项目”。这篇文章就说清楚我在做气象大数据处理时是怎么把分布式计算落地的从数据形态分析、存储设计、计算场景拆解到参数调优和踩坑记录全部是实操层面的东西希望能帮你少走一点弯路。文章适合三类人看刚接触气象大数据、正准备选型的数据工程师在气象行业做算法落地、被数据规模卡住脖子的开发人员以及学习了Spark/Hadoop但想知道“真实气象业务到底怎么用”的学生和研究同行。1. 项目整体设计与思路拆解气象数据到底该怎么喂给分布式集群1.1 气象数据的三座大山体量、时效、复杂度做任何分布式方案第一个动作不是写代码而是把数据形态摸清楚。我接触到的气象数据大体上可以归成四类。第一类是地面观测和探空资料。全国几万个自动站加上各种专业观测网最密的观测频率到分钟级单站单日可以产生上千条记录。这类数据的特点是“行多字段少”本质上是结构化时间序列非常适合列式存储和SQL分析。第二类是雷达资料。一部天气雷达体扫一次大约6分钟包含十几个仰角层的反射率、径向速度等产品。单部雷达一个体扫压缩后也有几十到几百兆字节。华东、华南、华北这些雷达组网密集的地区一次拼图要处理几十部雷达的数据而且偏好“分钟级”出图。第三类是卫星资料。静止卫星现在十分钟级别就能给出多个通道的全圆盘扫描加上极轨卫星的暗轨数据单日原始通道数据量大得吓人。处理链路长、算法链多、定标订正复杂属于典型的“数据工程和大气物理搅在一起”的难题。第四类是数值模式输出。全球模式的原始格点场单次输出经常是几百GB到TB级别一天两次甚至四次。再分析资料更是动辄几十年积累、PB级别的体量。这四类数据放在一起就是三个字大、快、杂。“大”是体量“快”是时效要求“杂”是格式和语义高度不统一。分布式计算的选型、存储设计、处理流程本质上是绕着这三座大山转。1.2 为什么传统单机方案扛不住以前做气象数据处理很多单位的老一套是MATLAB、NCL或者直接用Fortran/Python读文件跑完后画图。单机方案的问题不是“跑不动”而是“跑得很憋屈”。举一个很现实的数字十万个站点的逐小时观测一年就是大约8.7亿条记录存成Parquet大概占据几十GB。单机统计分析一次运气好几分钟运气差半小时。可一旦要把十年、二十年的数据全量重算一遍或者做小时级的滚动更新单机就会卡在两个瓶颈上一是内存装不下二是CPU只能有一个进程在算。分布式计算解决的就是这两个问题。把数据切到多台机器上每台只处理自己那一份再通过框架统一汇总结果这就是思想核心。用生活类比来说单机处理像是一个人搬一整栋楼的砖分布式是用一台叉车加上一支搬运队关键不是单块砖搬得有多快而是整个楼能同时开工。1.3 我最终选定的技术栈与架构思路结合气象业务的实际场景我采用的是以开源生态为主的这套组合HDFS做原始存储Spark做批量计算主力Flink/Kafka处理实时流HBase/Doris类系统服务高频查询ParquetZstandard作为分析层的统一文件格式。Python做算法验证Java/Scala写生产链路。这套组合的选型逻辑其实很朴素。气象数据虽然五花八门但绝大多数统计分析都是“先过滤、再聚合、最后出指标”的模式这正好是Spark的强项。而需要秒级响应的临近预报、雷电告警等场景交给流式计算更合适。至于存储原始文件一定得留着做算法迭代HDFS的廉价横向扩展能力是最稳妥的存档底仓分析汇入数据仓库层则用列式格式让后续查询尽量少扫描数据。这里要特别强调一句不要一上来就追炫技架构。先看你的数据量和时效要求够不够得上“必须上分布式”。如果只有几个GB的数据、批处理跑几分钟也能忍那上Kafka加Flink加Spark纯属给自己找运维负担。分布式计算的价值在于规模和时效不在于“用了就显得先进”。2. 数据接入与存储层所有计算的胜负手都在这里2.1 多源数据格式与解析要点气象数据的格式是出了名的“各自为政”。我列个表把最常见的格式和它们的解析方案一次说清楚。数据类别常见格式解释要点需要注意的坑地面、探空观测文本/CSV、BUFR、二进制表格BUFR用WMO编码表解析必须依赖官方码表库编码表版本不统一会导致字段错位数值模式输出GRIB1/GRIB2推荐用ecCodes库解析支持Spark内批量解码GRIB文件里同名变量的单位、层次可能变化卫星资料NetCDF4/HDF5用h5py/netCDF4库注意分块读取chunk设置不合理会导致随机读取极慢雷达体扫自描述二进制、NEXRAD等需要按雷达型号写专门解析器各型号的仰角层数、距离库长都不一样常规观测记录JSON/CSV流式解析schema3提前定义站点缺测值和0值常常混在一起这里最重要的经验是解析层必须独立成模块并且要加上“格式语义校验”。气象数据不像互联网日志少一个字段可能不影响大局但缺一个测站的降水量可能直接让一个区域的平均雨量失真。我习惯在解析后统一做三件事时间统一成UTC、经纬度统一成十进制度、缺测值统一编码成NaN并且单独记录哪些记录被修正过。这个“可溯源”的元数据意识后面能帮你省掉无数扯皮。2.2 存储组织一张分区表和一种列式格式接入层之后是存储。我在实践中反复调整最后形成了这样的组织方式原始文件直接落HDFS按“数据类型/日期/时次”分目录清洗和标准化之后的结果统一写成Parquet按“年/月/日”静态分区。为什么用Parquet而不是继续用NetCDF或者GRIB原因有三个。其一是列式存储对“只取几个变量做统计”的查询极其友好读十列数据不用把一百列都扫一遍。其二是压缩率高气象变量的数值范围相对稳定Zstd压缩下很多变量能有3到5倍以上的压缩率。其三是天生适配Spark/Flink的谓词下推和剪枝优化HDFS块级别的过滤能直接把扫描量砍掉大半。分区策略也很有讲究。时间分区是气象数据最自然的分区方式几乎所有查询都会带时间范围。但要注意分区粒度和文件大小的平衡分区太细会产生大量小文件搞得NameNode内存吃紧、任务调度开销大分区太粗又浪费剪枝能力。我一般会控制每个分区内文件在128MB到256MB之间。如果单日数据量太小就按周甚至按月分区。2.3 实时接入与批量回补的取舍气象数据里有相当一部分是需要尽快进入计算管线的。现在很多台站的分钟级观测是先通过消息队列进入流处理做质量控制、缺测告警再落盘到HDFS。但真实业务中最麻烦的往往不是“能不能实时”而是“回补怎么做”。观测设备会断传、网络会抖动、上游模式会延迟发布。等发现数据缺了一大块再重新灌一次如果管道不支持“重放”回补就是一场灾难。我给团队的硬性要求是所有接入作业必须是幂等的也就是同一份数据重复消费多少遍最终落库结果都一样。实现方式就是落盘前先查分区如果对应时次的数据已经存在且校验和一致就跳过存在但不一致就覆盖。别小看这条真实业务里一个月至少有几次回补没有幂等设计的时候每次回补都是加班之夜。另外流处理和批处理不要各做一套。推荐的做法是流处理只负责“快速出粗结果”比如异常报警HDFS落盘的原始数据仍然走批量Spark做精细处理。流和批共享同一套解析器和标准化逻辑只差计算引擎不同。这样既保时效又保质量。3. 核心计算场景拆解与实现细节3.1 场景一十万级站点数据的长时间序列统计先来一个最基础也最高频的场景把一万个国家级站、加上九万个区域自动站的逐小时观测做成按区域、按月的统计产品。这类任务的SQL化程度极高。比如算区域月平均气温和累计降水量SELECT region_id, month, AVG(t2m) AS temp_avg, SUM(rain) AS rain_sum, COUNT(DISTINCT station_id) AS station_cnt FROM obs_parquet WHERE dt BETWEEN 2024-01-01 AND 2024-12-31 AND rain 0 GROUP BY region_id, month这里有两个特别容易踩的坑。第一个是缺测值和零值自动站降水缺测通常编码为特定的负值或NaN但很多脚本为了省事统一填0结果“没有观测”和“没下雨”在后续气候统计里完全不是一个概念会把区域累计雨量严重做低。第二个是站点分布不均城市站密、山区站稀做区域平均时不能直接粗暴地平均所有站点最好按站点密度或网格权重加权。在分布式实现上这类均匀聚合任务其实很简单真正要留意的是两件事一是读出数据时就要过滤别把全年数据先算一遍再筛二是站点和区域维度的映射表要用广播变量下发别做全量join。区域映射表通常只有几千行广播下去每个executor本地查一下比网上的shuffle快了不止一个数量级。3.2 场景二格点数据的重投影与重网格化气候产品、模式预报基本都是规则的经纬度格点但气象服务经常需要把格点值插值到流域、行政区域或任意站点的多边形上。典型的操作是把0.1度的全球格点降水重网格化到一个城市的街区级网格。分布式实现思路很简单把目标网格的每个像元看成一个“任务”计算它覆盖了源格点的哪些点再做反距离加权或面积加权平均。但朴素写法有个致命问题如果直接用两个双层循环遍历源格点和目标像元复杂度是O(N×M)格点数一上亿直接卡死。正确做法是用空间索引把复杂度降下来。一种实际效果很好的方案是先给源格点打上网格编码——比如用固定的行列号——再对目标像元算出它可能相交的源格点行列范围只在这几行几列里做插值。在Spark里可以用“源数据按行列号分区目标像元按行列号join”的方式让每个分区的插值计算都只需要读取局部数据。我在实际项目中遇到过类似的业务把全球0.25度再分析格点资料重网格化到某流域的精细网格数据总量十几个TB。用这种“格点编码局部join”的方式300个核心的集群跑一次全量重处理大约一个半小时而最初用暴力交叉拉网跑了整整一天都没跑完。差距就在算法设计上分布式框架能并行但并行救不了低效算法。3.3 场景三雷达/卫星资料的网格融合雷达拼图是我觉得最体现分布式威力的场景之一。几十部雷达每部都在自己的极坐标里描述回波要把它们拼到一张经纬度或平面直角坐标的底图上需要做坐标变换、波束阻挡处理、多雷达重叠区优先级判断。笨办法是逐像素遍历底图对每个像素去找附近的雷达波束并插值。这个方案的问题在于底图像素动辄上百万每个像素都去遍历所有雷达计算量直接爆炸。分布式做法是在概率上做减法先把雷达覆盖范围外的底图像素全部排除再让每个网格点只去对应的几部雷达里取数。具体说可以把底图按格网切块每个块只接收与它空间范围相交的雷达数据。在Spark里这相当于一次“空间范围join”雷达体扫数据广播范围框每个块只匹配落在自己范围内的雷达。匹配之后每个块内部是相互独立的天然并行。这个方案的实际效果是华东区域雷达拼图从串行的十几分钟压缩到两分钟以内而且增加雷达数量后耗时几乎不增长因为每部雷达只影响局部块。对于卫星资料的融合类似的思路也成立。卫星扫描带和地形/云量做空间对齐时同样先做粗匹配再做细插值避免全局笛卡尔积。3.4 场景四极端天气事件自动识别与追踪再上一个复杂度更高的场景从逐小时的高分辨率格点资料里自动识别持续高温、暴雨团、大风事件并跟踪其移动路径。这类任务的本质是“时空连通域分析”。拿高温事件举例先计算每个格点是否超过本地阈值比如站点气候95百分位得到一张“是/否”的布尔格点场然后要把连续多个时次、空间相邻且都超阈值的格点聚成一团才算一次事件。单机版本用BFS扫描连通域数据量小没问题格点场一大就扛不住。分布式版本我做过两种路线。第一种是把三维时空场看成“帧-格点”的序列用窗口函数先做时间维上的连续标记再做空间维上的连通域聚合借助图计算框架对网格相邻关系建边跑连通分量算法。第二种更工程化因为气象事件的连通域跨度通常有限可以按经纬度切块扫描局部连通域再用边界重叠合并全局结果。第一种结果更严谨第二种实现更简单。这个场景最值得分享的教训是不要把算法复杂度藏在看似简单的UDF里。有人把单机BFS直接写进Spark UDF每个分区内仍然在串行跑大递归数据倾斜和栈溢出问题接踵而至。分布式环境下算法必须可切分、可合并连通域这类问题要么交给专门框架要么就老老实实设计分块合并策略。4. 分布式计算参数设计与实战调优4.1 资源规划先算账再搭集群搭集群之前先算一笔账这是很多团队容易忽略的。气象数据的处理往往是“高峰集中、低谷空闲”比如模式资料每天两次发布发布后的两个小时是全集群最忙的时候其他时间的负载相对平稳。所以资源规划不能只按平均负载算要按“高峰窗口能扛住”来预算。我给你一组参考数字。假设每天新增原始数据2TB压缩后落盘约500GB保留一年就是180TB左右加上存储副本和临时作业空间大约需要400~500TB的裸容量。如果调度高峰窗口要求2小时内完成全部批量处理按照每个CPU核心每小时能处理5GB压缩数据这个比较保守的经验值假设需要处理500GB大约需要20~25个核心在满负荷运行考虑到调度开销和重试实际准备40~50个核心比较稳妥。这个数字只是起算点。真正的决窍是预留弹性集群在私有环境里预留20%的空闲资源给临时探索性作业别把资源吃满。资源吃满意味着一个任务卡死就会拖垮后面所有任务这种连锁故障在气象数据批处理里非常常见。4.2 并行度、分区键与数据倾斜并行度是最常说也最容易被忽视的问题。Spark里有个老生常谈的公式每个分区的数据量控制在100到200MB之间分区数大致等于总数据量除以单分区大小。但气象数据有自己的特殊之处——时间分区天然存在热点盛夏对流天气多雷达、卫星数据量会突然比冬季大好几倍模式发布时点也会让某个小时的增量数据暴涨。所以做分区键设计时我建议不要只用单一时间字段。常用气象分区键组合有(data_type, year, month)、(region, day)、(model_run_hour, variable)。原则是让查询裁剪收益最大化和写入负载均匀化之间取平衡。加入region维度可以有效避免“全国数据堆在一个分区”但也要小心某些区域数据量天然偏大比如大城市周边站点密、数据多。数据倾斜在气象场景里特别典型。比如雷达拼图时某块格网正好覆盖对流单体那块的计算量会比其他块多几十倍再比如站点join区域表时人口密集省份的站点数远大于西部省份。解决办法有三板斧加随机盐破坏热点、改用范围分区代替哈希分区、把倾斜的join改成广播小表再做本地关联。我用得最多的是第三种因为气象业务里“大表join小表”场景很多小表广播出去后问题直接消失。4.3 Spark关键参数调优清单下面是我在气象数据批处理上反复验证过的一组参数基线可以直接作为起点。参数推荐设置选择理由spark.executor.memory4~8GB气象格点数据单分区通常不大给太大反而浪费GCspark.executor.cores2~4每个executor核心数太多会导致IO争抢spark.sql.shuffle.partitions数据量/128MB可调默认200对大表偏小容易产生大shuffle块spark.serializerKryo气象POJO字段多Java序列化太慢spark.memory.fraction0.75大部分场景是计算型任务给执行内存多留空间spark.sql.inMemoryColumnarStorage.batchSize10000适合中等宽度的气象表spark.sql.adaptive.enabledtrue自动合并小分区和调整join策略强烈建议开spark.sql.adaptive.coalescePartitions.enabledtrue解决小文件碎片化问题spark.sql.autoBroadcastJoinThreshold10MB~50MB气象维表一般很小放大阈值能少很多shufflespark.sql.parquet.compression.codeczstd压缩率比snappy高解压速度也能接受这些参数不是万能药但基本能覆盖气象数据“宽表少、时序列多、聚合重”的负载特点。一个基本原则先看是不是数据本身有问题再谈调参数。数据都倾斜成一边倒你把executor内存调到无限大也没用。4.4 一次压测的实际数据为了把上面的调优讲得不是空话我给出一次实际压测的模拟数据。数据是某模拟项目X的三年站点观测约2.6亿行记录Parquet压缩后约280GB集群规模为16节点、每节点32核、128GB内存实际可用Spark资源约128个executor核。压测任务为计算每个站点每月的平均气温、累计降水、有效观测天数。基线参数默认Spark设置、无自适应跑下来耗时47分钟把shuffle分区数调小、开启自适应执行引擎、列式压缩换成zstd之后耗时降到18分钟。再进一步把站点维表和区域维表广播join阶段几乎无shuffle整体耗时压到12分钟。同等工作负载用单机Python跑跑了近9个小时才出结果。这个对比不是炫耀数字而是想说明真正的收益来自“减少不必要的数据移动和序列化开销”。字段裁剪、分区剪枝、广播小表每一项单独看都是小优化合在一起却能差出近4倍性能。5. 常见问题与排查技巧实录5.1 问题速查表我在气象数据处理项目里踩过不少坑这里整理成一份速查表方便你对照排查。现象常见原因排查思路解决手段作业卡在某个stage长时间不动数据倾斜某个分区的任务量远大于其他看Spark UI里stage各task耗时分布加盐、范围分区、广播小表读取NetCDF/GRIB时偶发解不开文件损坏或库版本不匹配检查源文件的校验和确认解析库版本统一解析库版本做源文件完整性校验写入Parquet后查询结果与原始值不一致缺测值处理不统一、时区混淆对比源文件数值和结果表的count/极端值统一缺测编码和时间标准落表前加校验多个小文件让NameNode内存告急分区过细或写入并行度太高查看HDFS文件数和分区大小用自适应合并分区、按周/月归档collect()到driver直接OOM计算逻辑里非要用collect回收全量数据检查代码看有没有把大数据收回到driver改成分区聚合或limit抽样同一个变量两次计算得到不同结果数值模式不同时次的高斯格点定义不同检查格点表头里的网格参数在解析阶段写出每个变量的网格元数据Kryo报序列化错误POJO里的字段类型不兼容或未注册看栈顶字段名提前注册类、避免Map嵌套复杂类型流式任务重启后重复消费没有幂等设计检查offset提交逻辑和落库去重落库前判断分区是否已存在5.2 两个典型案例复盘第一个案例是“雷达拼图数据倾斜”。当时国产某型号雷达因强对流天气进入体扫加密模式单部雷达数据量膨胀到平时的5倍而它恰好覆盖了拼图底图的边缘两块。结果那两个块的计算任务是其他块的十几倍整个拼图作业被拖到超时。我们当时的修复分了两步第一步临时给那些热点块多加一层随机盐打散负载先把拼图时效保住第二步长期修复是把底图切块方案从固定经纬度切块改成按雷达覆盖密度动态切块热点区域切小一点、边缘区域切大一点。这个教训让我记住了气象数据“忙时更忙”的特征资源规划必须考虑极端天气下的峰值。第二个案例是“回补数据导致重复记录”。某次上游设备断传12小时修复后积压数据一股脑灌进来流式消费端在重启后从头消费了同一段Kafka消息导致HBase里出现了大量重复观测记录。事后查询统计时发现站点数对不上排查了好久才在明细表里看到时间戳完全相同但记录ID不同的数据。修复方式是在落表前加primary key去重并且给所有接入作业加了幂等标记。现在团队在处理任何接数任务时第一个问的问题永远是“重复消费了怎么办”而不是“先跑起来再说”。5.3 避坑心得这里再把压箱底的三条经验单独拎出来说。第一条气象数据的时间标准必须统一。UTC还是北京时间、秒还是毫秒、东八区还是零时区这些问题看起来蠢但实际项目里几乎每个月都会有人因为时区换算出错导致统计对不上。我强烈建议所有内部存储一律用UTC的Unix时间戳只在最终展示层转换成北京时间。第二条不要把气象算法和工程逻辑焊死在一起。气象算法是高度迭代的插值方法今天用反距离加权明天可能换成克里金质量控制阈值也会随季节调整。如果把算法逻辑埋在Spark的UDF里后续每改一次算法就要重新打包上线。更好的做法是把算法抽成独立函数库Spark只负责数据的切分、调度和结果汇总算法库即插即换验证也快很多。第三条一定要给“脏数据”留一条可追溯的通道。分布式处理会把原始数据拆散、变换、聚合一旦结果出问题如果没有原始档案可以回溯就只能干瞪眼。我在每一项写出的结果表旁边都会记录它的上游分区路径、解析库版本和生产作业版本。这样每次出问题都能快速定位是数据、代码还是环境引起的而不是靠猜。6. 结尾一些我踩过坑后换来的体会最后说点个人感受。分布式计算在气象数据处理里之所以有用不是因为它听起来高级而是因为气象数据天然就具备分布式计算喜欢的两大特征一是可以按空间、时间切分二是每个切分块上的计算逻辑几乎一致。只要抓住这两点方案设计就不会跑偏。我自己在实际操作中最大的体会是先把数据治理做好再谈计算引擎调优。什么格式统一、分区合理、缺测规范、可重放接入这些看起来不性感的工作决定了后面所有计算到底稳不稳。很多项目跑得慢、改得痛根子都在数据层面没有打好底子。这个内容后续其实还可以往两个方向扩展一个是把流式计算和批处理进一步统一用同一套表模型同时服务实时告警和离线统计另一个是往机器学习方向走把分布式特征计算和模型推理串成一条完整的天气智能预报链路。如果后面有机会我再把这两块的实际踩坑过程整理出来分享。