接手这个项目时我第一反应不是“又来个清洗统计的活”而是觉得天气数据叠加大数据框架这件事放在西南这种地形复杂的区域特别有意思。西南地区既有高原又有盆地海拔从几百米跨度到几千米天气现象的空间差异非常大。传统的气象分析工具处理小批量数据还行一旦面对几十GB甚至上百GB的历史站点观测数据和再分析资料单机方案根本跑不动Spark就成了最务实的选择。这篇文章我会把整个项目从数据接入、清洗、存储设计、分析计算到最后的可视化展示和模型应用的完整链路拆开讲。内容包括我踩过的坑、调参的教训、以及一些能直接“抄作业”的代码片段。无论你是刚开始接触Spark的数据分析师还是准备做类似区域气象研究的学生都能从中找到可以直接落地的思路。1. 项目定位与整体设计思路1.1 为什么用Spark而不是Pandas很多人问过我一个问题天气数据能有多大至于上Spark吗说实话如果只是分析某一天或某一个站点的数据Pandas确实够用。但这个项目要处理的是西南地区多年份、多站点、多要素的历史数据加再分析数据累计体量能达到几百GB而且里面还涉及站点观测和格点数据的融合。Pandas在这个量级下光加载数据就能把内存吃光更别说做复杂的聚合和关联计算了。Spark的优势在于它把数据切碎后分布到多个节点上并行处理内存不够就落磁盘单机搞不定的量级它可以扛住。更关键的是Spark SQL提供的DataFrame API让写分析逻辑的体验非常接近Pandas学习曲线没那么陡峭。我用的是PySpark主要原因是我们团队Python生态相对成熟后续接机器学习和可视化都方便。这个项目本质上做的是“降维”把分散的、格式混乱的原始天气数据加工成结构化、干净、可直接分析的数据表再基于这些表去回答具体的气象问题。Spark在这里承担的不只是计算引擎更是一个统一的数据加工厂。1.2 整体技术架构选型整个项目的技术栈并不复杂但每一层的选型都是经过反复权衡的存储层数据落在HDFS上文件格式统一用Parquet。列式存储对这类“按列聚合”的气象分析非常友好查询时只需要读取涉及的列IO开销能省一大半。计算层Spark SQL配合DataFrame API偶尔用RDD处理一些特别不规则的数据源。调度层用的Yarn做资源调度队列资源不够的时候任务会排队这个后面会讲到。结果层分析结果导出为Parquet和CSV两份Parquet供后续数据服务使用CSV给业务方做报表。至于为什么不选Hive做全部计算、为什么不用Flink核心原因就是Spark的批处理能力在这个场景下性价比最高。天气数据的分析大多是离线批任务按天或按小时跑一轮对秒级延迟没有要求。Flink更擅长流式场景Hive的交互体验又不如Spark SQL方便所以最终敲定Spark作为主计算引擎。1.3 西南地区天气数据的特殊性西南地区横跨多个气候带数据本身就有很强的空间异质性。比如同一时间下高原站点和盆地站点的气温可能相差十几度这种差异不是简单的均值能代表得了的。做分析时如果笼统地算全国或全区平均很多有价值的信息会被掩盖掉。所以我在项目里特别强调“按站点海拔地形区域”的维度去做切分。地形数据叠加气象数据后很多规律才真正显现出来——比如迎风坡和背风坡的降水量差异、山谷地带的夜间辐射降温效应等。这些特征如果不结合地理信息单纯看数字是看不出来的。另外一个麻烦是数据质量问题。西南有些站点位于偏远山区设备故障率高数据中断、跳变比平原地区严重得多。这意味着清洗环节不能只做简单的去空值还需要结合气象物理常识做合理性检查。2. 数据链路与预处理方案2.1 数据源盘点与接入方式这个项目涉及三类数据源每一类的接入方式都不一样气象站点观测数据从某气象数据服务商那里获取的历史站点逐小时观测资料包含气温、降水、风速、风向、气压等要素。单站点单要素全年大约8760个记录西南地区有数百个站点几年累计下来数据量并不小。再分析格点数据类似ERA5这类全球再分析数据集的西南区域切片分辨率是0.25度格点。这种数据不是“测出来”的而是通过数值模式同化后的产物适合用来补地面站点覆盖不到的地方。地形辅助数据SRTM高程数据用来提取每个站点的海拔和地形参数。这部分数据不需要频繁更新一次处理完入库即可。接入方式上站点观测数据是csv文件按站点分目录存储的我直接用Spark批量读取。再分析数据是netCDF格式PySpark不能直接读得先通过Python的科学计算库转为Parquet中间格式。这里有个小技巧netCDF转Parquet时最好一次性把经纬度和时间维度全部展开成列避免后面对行索引做笛卡尔积式的计算。2.2 数据清洗的三个关键坑天气数据的清洗是全项目中最花时间的环节远比后面的分析要耗神。我总结下来有三个坑必须提前做好准备。第一是缺失值处理。站点设备故障、通信中断都会造成数据缺失。空值太多的时候不能直接删行否则按小时汇总时会出现“某个小时没有数据”的假象。我的做法是如果一个站点某天缺测超过6小时那当天不参与日统计超过总记录5%的站点直接标记为低质量站点单独拉出来做敏感性分析。第二是异常值识别。天气数据有物理极限比如夏季气温不可能超过50度小时降水量不可能超过200毫米台风暴雨除外西南地区相对少见。我写了一个UDF按要素设定上下限范围超出范围的记录标记为可疑值再结合前后时刻的资料做线性插值替换或者剔除。第三是时间对齐。不同来源的数据时间基准可能不一样有的是世界时有的是北京时间相差8个小时。不统一时区会导致日最高气温和日累计降水全部错位。我先统一转成北京时间再按小时打齐时间戳这一步做完后整个数据体系才真正能串起来。2.3 数据分区与存储设计数据表怎么组织直接影响查询性能和开发效率。刚开始我图省事把所有数据都放一张大表里查询时用where条件过滤。结果是每次扫描都要读取全表跑一个简单的多年平均都要几分钟。后来我改成“按年和月双级分区”的物理存储结构即/data/weather_data/year2024/month07/每个分区里存的是当月的站点逐小时观测数据Parquet文件按站点做了排序。这样跑年度统计时只需要读取对应年份的分区跑月度分析时甚至可以只扫一个分区扫描量大幅下降。这里插一句小文件问题。天气数据的特点是“站点多、单文件小”如果每个站点每小时落一个文件HDFS上会堆出大量小于128MB的小文件NameNode压力很大Spark任务也会因为文件数太多而启动大量低效task。我的做法是每天做一次合并通过repartition(文件数, 按站点列)将当天的数据均匀分布到几个大文件中再写入Parquet分区目录。3. 核心分析场景与实现细节3.1 气温时空分布分析气温是最基础也是应用面最广的气象要素。我做了三块分析区域平均气温的长期趋势、月均温的空间分布、以及极端气温事件的识别。区域平均这块计算逻辑比较简单核心代码如下from pyspark.sql import functions as F df_daily df_hourly.groupBy(station_id, date).agg( F.avg(temperature).alias(avg_temp), F.max(temperature).alias(max_temp), F.min(temperature).alias(min_temp) ) df_region_month df_daily.groupBy(year, month).agg( F.avg(avg_temp).alias(monthly_avg_temp) ).orderBy(year, month)但这里有个统计口径问题求区域平均时站点分布是不均匀的盆地站点多、高原站点少直接算术平均会偏向盆地。更合理的做法是先按经纬度做网格化比如0.5度网格先求每个网格内站点的平均再对网格做面积加权平均。我最终用了简化版先按站点聚类到地形区域再在各区域内算平均最后加权合并。这个处理让区域平均气温的趋势分析更可靠。极端气温事件识别这块我采用阈值法加持续天数判断。比如连续三天日最高气温超过35度称为一次高温过程连续两天最低气温低于0度算霜冻过程。用Spark做这类“按站点分组、按日期排序”的连续过程识别时窗口函数是最顺手的工具from pyspark.sql.window import Window w Window.partitionBy(station_id).orderBy(date) df_temp df_daily.withColumn( hot_day, (F.col(max_temp) 35).cast(int) ) # 利用lag判断连续日期的起始点 df_event df_temp.withColumn( event_start, F.when(F.col(hot_day) 1, F.when(F.lag(hot_day, 1).over(w) 0, 1).otherwise(0) ).otherwise(0) )这个识别结果后来用在了农业服务端作为高温热害风险提示的底表。3.2 降水特征与雨季识别降水分析里最有价值的产出是“雨季起止日期自动识别”。西南地区降水季节性强雨季的起止直接影响农业生产安排。我用的方法是滑动平均法逐日降水量序列做5日滑动平均当年滑动平均序列第一次稳定大于某个基准阈值、且后20天没有明显回落的日期定义为雨季开始日。基准阈值不能全区域统一因为不同区域年降水量差异很大。我按站点历史平均年降水量分档年降水800毫米以下的站点阈值取3毫米/日800-1200毫米的取4毫米/日超过1200毫米的取5毫米/日。原理很简单阈值太低会把零星降水误判为雨季太高又会漏掉早期降水过程。区间累计降水、最大日降水这些指标用Spark的groupBy加agg就能完成。有一个容易被忽视的坑是降水的“微量记录”。气象资料里降水小于0.1毫米会记为0.0但对于降雨特征分析究竟是“完全无降水”还是“出现过微量降水”意义不一样。我建了一个辅助列区分“降水量是否为0”和“是否有降水现象”后续做降雨日数统计时全部基于这个辅助列避免口径混淆。3.3 风速风向与地形关系风速风向分析算不上这个项目的核心但做出来之后反而成了业务方最喜欢的内容。西南地区地形起伏大山谷风效应特别明显。我统计了站点平均风速的日变化曲线发现不少河谷站点凌晨风速小、午后风速迅速增大典型的山谷风日变化特征。为了把“地形风”的关系量化我在站点表里关联了海拔和地形粗糙度等级再按不同海拔区间统计风速均值。结果呈现出一个大致规律海拔越高、地形越开阔的区域平均风速越大盆地腹地站点风速明显偏小。这个结果在风电和电力设施规划场景里可以使用。风向处理也有讲究。不能用普通数值直接平均否则350度和10度平均出来是180度实际风向应该是0度左右。我自己实现了一个把风向转成单位向量的方法先求sin和cos的均值再通过atan2还原出平均风向角这样得到的才是气象学意义上的平均风向。4. 从分析到应用落地路径与效果4.1 可视化让数据自己“说话”分析结果如果只躺在表里价值很有限。我做了两层可视化第一层是常规统计图表用Python的Matplotlib和Seaborn库生成气温曲线、降水柱状图、风向玫瑰图第二层是空间分布图把Spark算出来的站点聚合结果导出成标准表格然后用GIS工具做插值和面状渲染最后输出成专题图。在空间分布图上站点稀疏问题会在插值时放大。比如高原西部站点少插值出来一个大面积的“无数据区”很容易误导人。我的处理是把插值范围限制在站点密度可支撑的区域同时把地形阴影叠加上去让读者直观理解“没数据的地方是因为站点覆盖不到”而不是真的没有气象现象。统一可视化之前的出口格式也很关键。我建议所有空间结果都带station_id、longitude、latitude、analysis_value四列就是标准的“点要素表”所有GIS工具都能直接读。不要在结果里混入多余描述列否则后面还要花时间做清洗。4.2 用MLlib构建气温预报模型分析完历史数据后我尝试用Spark MLlib建了一个简单的站点温度预报模型。这个模型的定位不是跟数值天气预报抢饭碗而是做“基于历史相似场景的快速预测”比如电网负荷预测需要下周每个小时的温度参考值这种场景下用一个轻量级模型随叫随用很方便。特征工程方面我选了海拔、月份、当日历史平均温度、昨日同时刻温度、三日滑动平均温度等几个变量。其中昨日同时刻温度是最重要的特征因为温度有很强的自相关性海拔用来捕捉不同站点的差异。标签列就是目标时刻的实际温度。模型选用了随机森林回归器Spark MLlib里的实现是RandomForestRegressor。不用线性回归的原因是温度与海拔、月份之间并非严格线性关系随机森林可以自动学到非线性结构。from pyspark.ml.regression import RandomForestRegressor from pyspark.ml.feature import VectorAssembler feature_cols [altitude, month, temp_lag1, temp_lag24, temp_ma3] assembler VectorAssembler( inputColsfeature_cols, outputColfeatures ) train_data assembler.transform(df_train) rf RandomForestRegressor( featuresColfeatures, labelColtemperature, numTrees100, maxDepth10 ) model rf.fit(train_data)最终模型在验证集上的RMSE大约2.3度在山区站点上误差会大一些高原站RMSE接近3.5度盆地站可以到1.8度。这个精度对业务参考来说足够用了。要注意的一点是用历史站点温度训练模型时训练集和测试集必须按时间切分不能随机切分。因为温度序列存在强自相关随机切分会让模型“偷看”未来数据评估结果虚高。4.3 应用场景拓展思路分析结果落地时我结合业务需求做了几个延伸场景农业气象服务利用积温指标对农作物适宜生长区做气候区划。项目输出的月均温、极端低温等结果结合GIS叠加作物种植边界可以识别出温度条件不匹配的区域给出调整建议。旅游舒适度评估用温湿指数和风效指数对重点旅游区域做舒适度月份分布生成的月度舒适度地图可以直接用于旅游线路规划参考。电力负荷预测气温与用电负荷有强相关性。项目输出的逐时温度序列作为外部特征接入某跨平台系统的电力负荷预测模型后预测误差降低了约12%。这些应用本质上都在做同一件事把气象数据变成可决策的信息。Spark在整个链条里负责提供“算得快、算得准”的基础数据产品。5. 常见问题与排查技巧实录5.1 数据倾斜的排查与应对我在跑“按站点聚合日降水”这个任务时发现整个作业特别慢观察Yarn日志后发现某个executor几乎承载了80%的数据量。原因是有几个中心站点的数据量比普通站点多一个量级按站点做groupBy时这几个热点键对应的数据全压在一个分区上。解决办法有两个层面。第一层是逻辑优化如果倾斜只影响少数键可以先做两阶段聚合先按站点加随机前缀做局部聚合去掉前缀再聚合一次。第二层是参数调整比如把spark.sql.shuffle.partitions调大让shuffle后的分区粒度更细同时spark.default.parallelism也相应调大。这种方式能缓解但未必根治根源在于数据本身的键分布不均衡。数据入库时如果能按站点做更均匀的哈希分桶后续的聚合倾斜会少很多。我在后续项目里把站点表改为分桶表按station_id分成64个桶聚合性能提升了近两倍。5.2 Executor OOM的解决过程有一次跑宽表关联任务时频繁报错ExecutorLostFailure原因是数据广播策略选择错误。一个大维度表几十万行我用默认的SortMergeJoin去和事实表关联导致每个executor都要维护一个较大的hash表内存直接被打爆。优化手段是改用广播变量把小维度表广播到每个节点from pyspark.sql import functions as F df_result df_fact.join( F.broadcast(df_dim), dim_key, left )Spark在内存中直接查找广播表不再需要shuffle。直播上做维度表关联时只要维度表小于Spark可用内存的阈值都建议优先用广播方式。另外把spark.sql.autoBroadcastJoinThreshold从默认的10MB调大到50MB能自动触发更多小表的广播优化。5.3 时间与坐标口径的统一问题时间口径问题在数据量小的时候不太明显一旦做跨年汇总就很容易出错。我第一次算“全年累计降水量”时发现结果比气象部门发布的参考值少了约百分之八。排查到最后发现是有三个站点的源数据用了UTC时间按日划分时记录落错了日期导致边界天的降水被漏掉。解决办法是清洗阶段就统一做一次to_utc_timestamp或from_utc_timestamp的转换并且对每天的数据做完整性校验检查每站点每天的记录条数是否等于24小时。不满足的站点标出来单独复查这个校验逻辑让后续所有分析的准确性有了基本保障。坐标方面同样有坑。部分格点数据用的是经纬度浮点表示但不同来源数据的浮点精度不同直接做空间匹配时对不上的概率很大。我先统一转成LonLat坐标然后把经纬度字段格式化成固定小数点位数再建一个“网格ID”字段后续匹配全部基于网格ID做等值关联又快又准。5.4 小文件与NameNode压力还有一个容易被忽略的问题就是数据生产路径里Parquet小文件数量失控。有一段时间天气数据服务端每15分钟生成一个增量文件当天合并前HDFS上会有上百个碎片文件。虽然Spark能读但每次列取任务光打开文件列表就要花几十秒。解决方案是建立一个每日合并的定时任务每天凌晨读取前一天的全部分区数据通过repartition按站点重分区目标大小控制在128MB左右然后覆盖写入。跑完后再执行OPTIMIZE做文件合并和气层统计更新确保后续查询使用最新的文件布局。写在最后的实操体会做天气数据分析项目最大体会是“数据加工流程比算法模型更影响成败”。很多人一上来就急着调模型参数结果发现数据里有大量脏点和口径不一致的问题后面再怎么调都是白费。扎实的清洗与存储设计是Spark这类项目真正能出活的前提。一个小技巧分享给大家在做任何正式统计分析之前先跑一遍全量数据质量报告分站点输出缺失率、异常值占比、时间连续性等指标。这个报告本身只要几十行Spark代码就能生成但它能帮你提前发现数据链路里的隐患省下后面反复排查的几天时间。另外涉及到空间分析时多花点时间把坐标系和网格ID的口径定清楚后面做插值和匹配会事半功倍。这个项目后续可以考虑的方向是接入实时数据流配合Spark Structured Streaming做逐小时的快速更新让分析结果从“离线报表”进化成“准实时服务”。到那个阶段基于天气数据能做的事情就更灵活了。