Spark处理两百GB全国气象数据：从清洗到性能调优的完整实战

简介：大数据处理中，传统单机Pandas面对数十GB乃至上百GB的结构化数据时，往往因内存瓶颈而无法胜任。Spark作为分布式计算引擎，通过内存计算、分区读取与Catalyst优化器，为海量表格数据提供了高效的分析方案。在实际工程中，数据清洗是保证分析质量的关键，包括缺测值映射、异常值过滤及质量控制，并将清洗结果转为Parquet列式存储以提升查询性能。同时，合理设置分区数、使用广播变量优化JOIN、调整序列化方式等性能调优手段，能显著提升Spark作业的稳定性与效率。本文以全国两千多个气象观测站、七十余年的气候记录为实例，完整展示了从技术选型、数据预处理、核心统计到可视化交付的实战路径，为处理大规模气象数据或类似多维表格数据的开发者提供一套可复用的参考方案。 去年接了个数据处理项目，甲方扔过来好几块硬盘，打开一看，是二十多个省份、两千多个气象观测站、七十多年的逐日气候观测记录。压缩包全部解压完大概两百多GB，我第一反应是这玩意儿用单机Pandas想都不用想，光读一遍就能把内存干到爆。所以我直接用Spark搭了一套分析流程，配合Python做图形化输出，前后一个月把整个流程从数据清洗、核心统计、可视化到结果交付完整跑通，中间踩了不少坑。

这篇文章我想把整个过程从技术选型、数据清洗、核心分析代码到性能调优完整记录下来，给准备用Spark处理大规模表格数据、或者正打算做气象/地理类数据分析的同学一份可以直接抄作业的参考。文章里涉及的都是我在实际项目里验证过的东西，没有什么“最佳实践”式的空话，基本都是线性的真实操作步骤。

1. 为什么用Spark做气象数据分析——技术选型实录

1.1 先算算全国历史气象数据到底有多大

很多人对“大数据”这个词没有概念，觉得几个GB的CSV也敢叫大数据，实际上拿到全国历史气象数据之后，你会发现这个量级远超单机内存能承受的上限。

先做一个简单的估算：全国大约有2400多个国家级气象观测站，如果使用逐日数据，从1951年算到2023年，一共73年，每年365天左右，那数据量大约就是：

2400 × 73 × 365 ≈ 6400万行。

但实际项目里拿到的数据往往不止国家级站点，还有区域级自动站，站点数量能到几万个，时间粒度也可能细化到逐小时。如果是逐小时数据，那行数直接乘以24，大约15亿行。再加上温度、降水、风速、湿度、气压、日照等十几个要素列，每条记录按100字节算，逐日数据大约6~7GB，逐小时数据超过150GB。

这还只是原始文本。如果换成未经压缩的文本格式加上各种分隔符、表头、换行符，整体体积还要膨胀30%~50%。我自己拿到的那批数据解压后是两百多GB，里面覆盖了逐小时和逐日两种粒度，加了质量控制标记字段，所以比估算值还大一些。

这种量级的数据，单机Pandas不是“慢”的问题，而是读不读得进内存的问题。64GB内存的机器一次性加载两百多GB原始文件，直接OOM，没有任何悬念。所以第一步决策就是把处理引擎换成Spark。

1.2 Pandas、Dask、Spark，我为什么选了Spark

在处理大数据表格时，通常有三条路：Pandas、Dask、Spark。

Pandas单表哪怕只有一亿行，在普通服务器上也已经很难玩得转了，groupby、merge这类操作一旦触发中间结果膨胀，内存会瞬间翻倍。Dask可以在单机上做并行计算，也能处理超过内存的数据，但它的生态和Spark相比还是薄弱一些，尤其是SQL优化、JOIN优化、窗口函数方面，性能差距明显。Spark的DataFrame API本身是模仿Pandas的，Python端的程序员上手几乎没有学习成本，同时它背后有Catalyst查询优化器，能对逻辑计划做谓词下推、列剪枝、分区裁剪，这部分优化是Dask目前比不了的。

另外一个关键原因是项目后期需要做全国维度的聚合分析，比如按年份聚合全国平均气温、按站点统计极端高温天数、按月份算降水量分布，这类操作本质上都是分组聚合和窗口计算，正是Spark最擅长的场景。

还有一点，纯粹从数据分析角度来说，Spark的SQL能力非常成熟。我可以直接把DataFrame注册成临时表，用SQL写聚合逻辑，这比用Pandas链式写法直观很多，尤其在处理多重group by和窗口函数时，SQL的可读性优势非常明显。

1.3 Spark在这个项目里具体解决了什么问题

我总结一下Spark在这个气象项目里实际发挥的作用，主要有四个方面：

第一，分布式存储和读取。数据源放在HDFS或者本地文件系统，Spark按分区读取，不会把所有数据加载到一个节点上。

第二，内存计算与迭代优化。气象分析经常会做多阶段聚合，比如先按站算月均值，再按月算年均值，最后在全国层面做趋势拟合。Spark的RDD和DataFrame会把中间结果缓存在内存里，避免每次聚合都重新读一遍磁盘。

第三，容错机制。跑一个几十亿行的任务，中间任何一个节点挂了，在MapReduce里可能整个Job重跑，Spark基于Lineage血缘关系，只重算丢失的分区，这在长时间运行的任务里太重要了。

第四，生态衔接。Spark计算结果可以非常方便地转成Pandas DataFrame，也可以直接写Parquet、CSV、数据库。我最后就是用toPandas把聚合完成的小结果集拿回本地画图，完全够用。

2. 数据清洗与预处理：把一堆原始文件变成可用数据

2.1 原始气象数据长什么样

气象数据的原始格式没有统一标准，常见的有两种来源。一种是国内气象数据网下载的地面气候资料日值数据集，一般是一个站点一个文件，文件名带区站号，内容是固定的列格式，每天一行，包含日期、平均气温、最高最低气温、降水量、风速、日照时数等。另一种是NOAA的GSOD/ISD数据，按年打包，全球站点都在一起，带有经纬度和质量控制字段。

两种格式都得先做解析。国内数据集最坑的地方在于缺测值往往用特殊数字表示，不同版本的数据集缺测值还不一样。我常见到的就有32700、32744、32766、99999等，这些都是历史数据里约定俗成的缺测编码，没有统一的规范文档，只能对着数据说明手工核对。

还有一个常见问题是站点迁站。同一个区站号对应的经纬度在不同年份可能不一样，如果不处理，后续做地理可视化时会出现站点漂移的诡异现象。

所以我把清洗流程设计成三个阶段：

1. 解析阶段：把不同来源的文件统一读成DataFrame，列名规范成英文小写加下划线，日期解析成标准timestamp类型。
2. 过滤阶段：缺测编码全部映射成NULL，质量控制码非0的记录直接过滤，异常物理值剔除。
3. 落地阶段：清洗好的数据重新分区，写成Parquet列式存储，后续所有分析都从Parquet读取。

2.2 清洗逻辑和关键编码规则

清洗过程看起来简单，但细节决定成败，我把几个关键点单独拿出来说。

第一个是缺测值映射。国内日值数据集的缺测值通常有多个，例如32744表示缺报或未观测，32766表示降水微量或者没有降水，在降水要素里还分0和32700两种语义。这些值如果直接参与聚合运算，会把平均气温拉低十几度，结果完全失真。我的处理办法是写一个UDF，把这些特殊值统一替换成NULL，之后再聚合时用avg函数自然跳过NULL。

第二个是异常值过滤。气温数据的物理合理范围大致在-55℃到50℃之间，降水的物理合理范围是0到400毫米日降水量，风速一般不超过75米/秒。超过这些范围的值要么是仪器故障，要么是录入错误，直接删掉即可。

第三个是质量控制码。部分数据集每行都有QC字段，0表示数据正确，1表示可疑，2表示错误，3表示已修改。我保留了0和3，过滤掉1和2，这个策略可以最大程度保留有效数据的同时把噪声降到最低。

第四个是站点信息表。我会单独建一张站点维度表，包含站点ID、站名、省份、经度、纬度、海拔，并处理迁站情况——如果一个站点的经纬度在不同时间段变化很大，我会保留最新的经纬度，并在分析维度上把它视为同一个站点。

2.3 从原始CSV到列式存储Parquet

清洗完的数据不会直接在原始格式上反复读取，而是统一转换成Parquet格式落地。

为什么选Parquet？三个原因：列式存储、高压缩比、内置统计信息。气象分析经常会用时间和站点做过滤，比如只看2020年以后的数据，或者只看某一省份的数据。Parquet的谓词下推再加上分区裁剪，可以让Spark在读取阶段就跳过大量无关文件，实际测试下来，查询速度比直接读CSV快三到五倍。

当时我用Spark读取清洗后的数据，按“年份”字段做了动态分区写入，每个年份一个分区目录。这样后续分析例如算年度趋势时，Spark只需要扫描对应年份的分区，不需要全量扫描整个表。

写到Parquet时的编码也很关键，因为原始数据里如果有中文列名，Parquet schema的兼容性会有问题，所以我在清洗阶段统一把列名改成了英文。时间字段建议存储成timestamp类型，不要存字符串，这样后续过滤效率高很多。

3. 核心分析代码与实现细节

3.1 气温趋势分析：年度聚合与线性拟合

分析任务里最核心的一个指标是“全国年平均气温变化趋势”。实现思路不复杂：从日值数据里按年份聚合出年平均气温，再用线性回归看斜率。

第一步是读取Parquet数据：

from pyspark.sql import SparkSession from pyspark.sql.functions import col, year, avg, count spark = SparkSession.builder \ .appName("weather_analysis") \ .config("spark.sql.shuffle.partitions", "200") \ .config("spark.sql.parquet.enableVectorizedReader", "true") \ .getOrCreate() df = spark.read.parquet("hdfs:///user/weather/clean_data/")

第二步按年份聚合全国平均气温。这里有个细节，不同站点的日值缺失情况不一样，如果直接对所有站点做简单平均，某些年份缺测站数特别多，平均结果的代表性会变差。所以我一般会要求有效站点数必须超过全国站点数的三分之二才认为该年均值有效。

yearly_temp = df.filter(col("avg_temp").isNotNull()) \ .groupBy(year("date").alias("year")) \ .agg(avg("avg_temp").alias("nation_avg_temp"), countDistinct("station_id").alias("station_count")) yearly_temp = yearly_temp.filter(col("station_count") > 1600) \ .orderBy("year")

第三步，把聚合结果拿到本地做线性拟合。因为到这里数据量已经非常小了，只有几十行，用toPandas拉回本地完全没问题。

import pandas as pd import numpy as np pdf = yearly_temp.toPandas() x = pdf["year"].values y = pdf["nation_avg_temp"].values slope, intercept = np.polyfit(x, y, 1)

算出来的斜率就是每年平均升温多少摄氏度。这个数字如果要用于正式报告，最好再算一下置信区间，我用的是scipy的linregress函数，可以同时给出p值和标准误。

3.2 极端天气事件统计：高温日数和暴雨日数

除了平均气温，极端天气的频次变化也是气象分析里非常关心的指标。这里我做了两个典型的统计：高温日数和暴雨日数。

高温日数的业务定义是日最高气温≥35℃记为一个高温日，≥40℃记为强高温日。我按年、按站点把高温日数统计出来，再汇总成全国或者分省的总量。

hot_days = df.filter(col("max_temp") >= 35) \ .groupBy(year("date").alias("year"), col("province"), col("station_id")) \ .agg(count("*").alias("hot_day_count"))

暴雨日数的定义是日降水量≥50mm记为一个暴雨日，≥100mm记为大暴雨日。

rain_storm = df.filter(col("precipitation") >= 50) \ .groupBy(year("date").alias("year"), col("station_id")) \ .agg(count("*").alias("storm_day_count"))

这类统计的坑在于降水要素和气温要素的缺测判断方式不一样。降水如果为0表示无降水，是正常观测值；而气温的0度是正常值，不能过滤。所以清洗时不能一刀切把所有0值都删掉，必须按要素区别处理。

再往深一层做，可以统计高温日数随年份的变化趋势。做法和前面气温趋势类似，把每年全国高温总日数或者平均单站高温日数做线性回归，看斜率为正还是负。这部分结果通常比单一平均气温更有说服力，因为极端天气的频次变化对公众感知更直观。

3.3 站点维度的精细化分析：窗口函数与滞后计算

如果要看更精细的规律，比如单站点的气温变化率、季节转换时间的变化，就需要用到窗口函数。

比如我想计算每个站点的年平均气温在时间上的变化，可以用lag函数计算相邻年份之间的温差：

from pyspark.sql.window import Window w = Window.partitionBy("station_id").orderBy("year") station_yearly = df.groupBy("station_id", year("date").alias("year")) \ .agg(avg("avg_temp").alias("year_temp")) station_yearly = station_yearly.withColumn( "prev_year_temp", lag("year_temp").over(w) ).withColumn( "temp_change", col("year_temp") - col("prev_year_temp") )

这样每个站点每年相对于上一年的温差就出来了。汇总统计时如果把temp_change按年做平均，可以看出整体升温在哪些年份最明显，哪几年有降温波动。

窗口函数的坑是如果不指定partitionBy，默认按全表排序，那会把所有数据放到一个分区里算，压力极大。气象数据分析时分区键一定是站点或者年份这种分布均匀的列，而不是温度这种连续值。

再举个例子，30年气候平均值的计算也适合用窗口函数实现。所谓气候平均，一般取最近30年，如1991-2020年平均。用窗口函数按站点和月份分组滑动30年窗口算平均，比先过滤再聚合高效很多，因为不需要反复扫描整个数据表。

4. 可视化落地与业务洞察

4.1 什么时候把Spark数据拉回本地

很多初学者拿到Spark就喜欢把数据collect回Pandas再处理，这个思路在数据量大时是致命的。

Spark计算结果分为两种：一种是聚合后的结果，比如年均温度，每年全国就一行，这种数据量很小，用toPandas拉回本地画图完全没问题；另一种是全量明细数据，比如所有站点每天的原始记录，几百GB，这绝不能拉回本地。

我的原则是：所有分析都在Spark里完成聚合，最终拉回本地的必须已经是“结果表”。比如画全国气温趋势图，我只需要年份和对应平均温度两列，最多几十行；画站点分布图，我需要每个站点的经纬度和某种度量值，大约几千行。几千行的DataFrame用toPandas一点压力都没有。

画图部分我用的是Matplotlib加Seaborn，地理分布图用Cartopy画中国地图边界，然后再叠加散点图。

4.2 趋势图、热力图、分布图怎么画

整个项目里我生成了三类核心图表。

趋势图是最直观的。把年份作为x轴，平均气温作为y轴，画折线图，再叠加一条线性回归直线，能一眼看出升温趋势。这里有一个视觉优化细节：折线图的线宽调到2.5左右，回归线用虚线加粗，颜色错开，不然两种线容易混在一起。

热力图适合看月份-年份维度的气温变化。把数据透视成年份×月份的矩阵，用Seaborn的heatmap画，暖色表示高温，冷色表示低温。不过从视觉可读性出发，我建议还是按年月折叠成趋势曲线更直观，热力图只适合做展示大图，不适合做核心结论。

分布图用于看地理特征。把每个站点的经纬度和指标值（比如年均温变化斜率）组合起来，在地图上画散点，点的大小和颜色映射指标数值。这里必须注意地图投影坐标系要和散点经纬度坐标系一致，我使用的是EPSG:4326，即WGS84经纬度坐标系，Cartopy的ccrs.PlateCarree就是匹配这个的。

4.3 从数据里看到的几个有意思的规律

分析结束之后我总结了一些有价值的业务发现，这里挑三个有代表性的简单说一下。

第一个是增温速率的空间差异。从单个站点的气温变化斜率来看，我国北方大部分站点年均温升高速率高于南方，具体数字这里不展开，但从业务角度看，这意味着不同区域的气候适应策略应该差异化，而不是一刀切。

第二个是极端高温日数的年际波动明显增大。高温日数不是线性增长的，而是呈现所谓“跳跃式增长”，某些年份突然暴增，之后维持在较高水平。用滑动平均处理后会看得更明显。

第三个是暴雨事件的不均匀性。暴雨日数在南方站点显著高于北方，但近年来的趋势是北方部分站点的暴雨日数也在增加，这种区域性差异从聚合统计里很容易发现，但从单一站点看很难看清全局。

这些规律是整个项目的核心交付物，也是Spark这种全量分析工具的价值体现——不用抽样，直接跑全量数据，结论的置信度完全不一样。

5. 性能调优与集群部署踩坑记录

5.1 分区数和并行度怎么设

Spark的性能调优，第一刀永远是看分区数和并行度。

默认的spark.sql.shuffle.partitions是200，这个值对几十亿行的数据明显不够。我一般会根据executor数量和CPU核心数来估算：分区数设为executor总数乘以核心数再乘以2到3倍。

举个例子，我用的集群是8台机器，每台8核，那总核心数是64。目标并行度我设置在128到192之间。设置方式：

spark = SparkSession.builder \ .config("spark.sql.shuffle.partitions", "160") \ .config("spark.default.parallelism", "64") \ .getOrCreate()

还有一个细节：读取Parquet时每个文件最好在128MB左右，太小的文件会有大量task空转，太大的文件则容易让单个task处理时间过长。写Parquet时如果发现小文件过多，可以先repartition或coalesce再写。

5.2 广播变量用得好，JOIN速度翻倍

气象分析经常要把大表和站点信息表做关联。站点信息表全国也就两千多行，这种“大表join小表”的场景，直接使用broadcast join会产生极大性能收益。

默认情况下Spark的join如果遇到一张小表，会自动广播，但保险起见，我会显式指定：

from pyspark.sql.functions import broadcast df = df.join(broadcast(station_info), "station_id", "left")

广播之后，小表会被复制到每个executor的内存里，整个join过程不需要shuffle，性能提升立竿见影。

这里需要注意广播变量的内存限制。默认的广播阈值是10MB，如果小表超过这个值，要调大参数：

.config("spark.sql.autoBroadcastJoinThreshold", "104857600")

5.3 Shuffle与序列化配置经验

Shuffle是Spark作业里最昂贵也最容易出问题的环节，尤其是groupBy和join操作。我调优时主要关注几点。

第一，序列化方式。Java自带的序列化器性能比较差，我换成Kryo：

.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

Kryo序列化后的数据更紧凑，网络传输和磁盘溢写的压力都会小很多，实测性能提升20%以上。

第二，Shuffle落盘。如果executor内存不足，Spark会把shuffle数据溢写到磁盘。这时要关注spark.shuffle.spill.compress配置，默认是true，一般保持默认即可。如果发现溢写严重，说明executor内存不够或者分区数太少，优化方向是增加shuffle.partitions。

第三，Executor的内存分配。每个executor分配的内存不宜过高，太高会导致GC压力增大。我试过4核8G的executor配置，比较稳定。堆内内存和堆外内存的比例用默认值就行，不需要特别调。

第四，缓存策略。同一个DataFrame如果会被多次使用，用cache或persist缓存在内存里。气象数据聚合之后经常会被多个分析模块复用，比如“日值数据”这个表，我先做了一次清洗缓存，再分别计算趋势、极端天气、月度统计，节省了大量重复IO。

5.4 本地调试与集群运行的差异

本地跑Spark和集群跑Spark，体验差异非常大，很多坑都是本地不出问题，一上集群就挂。

本地模式下，Spark默认把driver放在当前进程，所有任务都在本地线程池执行，executor内存默认是1GB。数据量大一点就会报OOM。所以本地调试一定要限制数据量，我通常是先filter某一年或者某个省份的小样例数据跑通逻辑，再在集群上全量铺开。

还有一个常见问题是Python版本和依赖不一致。Spark的PySpark执行Python代码时，会通过pyspark.daemon在executor上启Python进程，如果每台worker的Python环境缺少依赖包，任务会在运行中途失败。解决办法是在每台机器上装好相同的依赖，或者用spark.pyspark.python指定一个统一的Python解释器。

6. 常见问题速查与排障实录

6.1 高频问题排查表

我把这个项目里遇到的高频问题整理成一张表，后面做类似项目可以直接对照排查。

6.2 一次典型的OOM排查过程

我印象最深的一次OOM排查发生在算全国极端降水日数的时候。那次任务跑了一个多小时，在某个stage反复失败，错误信息显示executor lost，每个executor都报OutOfMemoryError。

我的排查步骤是先看Spark UI里的executor内存使用曲线，发现shuffle read数据量异常大，说明shuffle.partitions设置不够，导致每个task拉取的数据过多。我先把shuffle.partitions从200调到500，问题缓解了一些，但还有两个executor在崩溃。

再进一步看数据，发现降水数据的groupBy key是“省份+年份”，但部分省份的站点数量远多于其他省份，比如站点特别多的省份那个task要处理的数据量是其他省份的几倍。这属于典型的数据倾斜。

最终解决方案是用两阶段聚合：第一阶段先按“站点+年份”聚合出每个站点的暴雨日数，第二阶段再按省份汇总。这样第一阶段的shuffle key是站点ID，分布均匀，第二阶段的中间结果已经大幅减小，不会再产生倾斜。

这个案例的启示是：遇到倾斜问题优先从业务逻辑层面拆解，先把细粒度的中间结果聚合出来，再用上层维度汇总，往往比直接在最终维度上做groupBy要快一个数量级。

6.3 运行环境与资源规划避坑建议

最后补充几个关于运行环境的建议。

我这次用的是8台机器的小集群，每台机器64GB内存、16核CPU。Spark部署用的是Standalone模式，没有上YARN，主要是因为集群私有，不需要多租户资源隔离。

如果你手头没有集群，也可以在本机用Spark的local模式跑这个项目，不过需要把数据量缩小到一个合理的范围，比如先只处理某几年或者某几个省的数据。想跑全量的话，建议把clean数据做成分区Parquet，然后至少准备一台32GB内存以上的机器，local[*]模式也值得一试。

另外一个建议是：在大规模作业跑之前，先用小的数据集跑一遍逻辑验证，再上全量。这里面最基本的做法是用limit(1000)或者year过滤来做开发测试，不要一上来就全表扫描。开发环境和生产环境的分离，我吃过不少亏，这一条值很多时间成本。

写在最后的一点体会

这个项目跑完之后，我最大的感受是Spark真正解决的是“单机装不下、跑不动”的硬性诉求，而不是为了用框架而用框架。实际上最后交付的分析报告核心图表也就十几张，但每一次聚合计算都建立在几千万甚至几十亿行的全量数据之上，这种结论的说服力是完全不一样的。

如果你手头正好有一批类似的多维表格数据，我的建议是先在本地用一个小样例跑通逻辑，再放到Spark上铺开。清洗环节宁可多花点时间把缺测值和质量控制搞清楚，也不要急着做分析，因为气象分析这种领域，数据本身的准确度比分析技巧重要得多。踩过几次坑之后你就会明白，数据清洗阶段每多花一小时，后面分析阶段就能省下十几个小时。

本文还有配套的精品资源，点击获取