★ FEATURED ARTICLE 简介一套基于Spark的全国历史气象数据分析项目代码完整、运行配置说明齐全面向计算机相关专业学生、教师、科研工作者及行业从业者适合作为毕业设计、课程设计、作业或项目初期立项演示也可以在其基础上继续开发新功能。资源包共75个文件大小2.46MB包含5个Python脚本如气象站点入库、年度统计、全国地图与曲线可视化等35份Markdown说明与开发记录txt格式的气象站点和历年数据文件7张全国最高/最低/平均气温及降水量分布图、气温与降水变化曲线图以及答辩PDF和附赠内容压缩包文件类型覆盖源码、数据、文档、图表与演示。结合目录与文档可按模块学习Spark数据处理、Hive/MySQL数据落地和Matplotlib可视化流程从站点数据处理到全国指标计算均有对应代码可对照。已有29人学习下载对想以真实气象数据练习Spark编程和数据分析的读者有直接参考价值。1. 为什么这个项目值得用Spark跑一遍全国两千多个国家级气象站每天一份观测记录二十年的历史气象数据攒下来就是千万行级别。我第一次用Pandas清洗这批数据第二轮聚合内存直接顶不住换到Spark后同样的分析流程几分钟跑完。所谓“Python基于Spark的全国历史气象数据分析项目”就是写PySpark作业把气温、降水、气压、湿度这些要素读进Spark完成清洗、重采样、统计和可视化最终产出全国趋势图、极端事件统计表这类可验证的结果。它适合正在学Python数据处理、想拿真实spark数据分析案例练手的从业者也适合气象、能源、农业方向需要批量处理站点数据的同学。这套技术栈换到农产品价格数据分析类场景套路基本一致只是换一套字段和阈值而已。2. 环境搭建版本匹配是省时间的第一件事2.1 为什么本地跑也要用Spark规模、惰性求值和迁移成本很多人一听到Spark就想到集群觉得本地开发用不上。实际做全国历史气象数据这个量级Spark的价值不在于多机并行而在于“分区处理 惰性求值”这套计算模型。Pandas在处理千万行数据时不是跑不动而是每一轮清洗逻辑调整都要全量重算改一次条件、加一个字段前面加载和转换步骤跟着重来一遍开发效率太低。网上的spark数据分析案例大多是做一个join再做一个groupBy真实的气象项目第一道坎反倒在文件形态和脏数据上。PySpark的DataFrame API和Pandas、SQL的习惯很接近写起来不像换个框架你能继续用select、filter、groupBy这些熟悉的概念只是计算被分散到多个分区上并行执行。这套迁移成本很低值得一开始就进入Spark语境而不是先用Pandas写一版再回头改。还有一个新手最容易卡壳的思维切换Spark的变换是懒执行的。你写filter、withColumn这些步骤时Spark只是在构建计算图没有真正读数据只有遇到count()、show()、write这类action操作任务才会真正触发。所以调试时不要每改一行就跑一次全集先用小文件或limit切片验证逻辑。2.2 安装与最小验证Python、PySpark、SparkSession先确认本机Python环境。我习惯用3.8及以上的稳定版本如果你连Python都没装好先按官网的python安装教程装一个这里不展开。PySpark直接通过pip安装国内环境建议带镜像参数否则下载慢而且容易超时python --version pip --version # 使用清华镜像安装 pyspark pip install pyspark -i https://pypi.tuna.tsinghua.edu.cn/simple装完之后第一步不是急着读数据而是确认SparkSession能正常创建、Driver和Executor能通信。下面这段是每次新环境我都先跑一遍的“最小验证”from pyspark.sql import SparkSession # 本地模式local[*] 表示用满本机所有 CPU 核 spark SparkSession.builder \ .appName(weather-min-check) \ .master(local[*]) \ .getOrCreate() # 一个最基础的分布式求和把 4 个数字分发到分区再汇总 sc spark.sparkContext result sc.parallelize([1, 2, 3, 4]).sum() print(result , result) # 期望输出 result 10这里有两个参数值得说清楚。master(local[*])是指以本地模式运行星号表示用满本机CPU核如果你只想用两个核做调试可以写成local[2]这样能模拟并行度较小的环境。appName只是给这个作业起个名字在集群模式提交时会在资源管理界面上看到它本地模式不起实质性作用。如果打印出来的结果不是10说明底层通信有问题先解决环境再继续不要带着隐患往下写。Windows本地跑Spark还有一个特有的坑启动时如果报错提示找不到winutils相关二进制需要去配置HADOOP_HOME并放入对应版本的winutils可执行文件。macOS和Linux一般没有这个问题。这个坑和你的代码无关纯粹是Hadoop在Windows平台缺少本地库导致的配置好环境变量就能过。2.3 本地模式先行先别急着搜集群搭建教程本地模式足以支撑千万行级别的气象数据分析和调试所以我的建议是先别急着搜spark集群搭建教程把local模式跑通再说。网上大量spark集群搭建教程讲的是YARN或Standalone集群配置涉及多机部署、资源规划这些在开发阶段用不上只会增加变量。真正需要上集群的信号很明确数据量到十亿级以上、需要定时调度、或者有多人共享计算资源。到了那时你的PySpark代码基本不用改只是把提交方式从“IDE里直接跑”换成spark-submit --master yarn。你在网约车大数据综合项目里看到的Hive离线分析套路本质上是把SQL翻译成分布式任务用Spark SQL重写一遍也是同样的效果但不需要搭Hive那套重量级组件。工程目录一开始就按可扩展的方式组织我习惯这样放D:/weather_project/ ├── data/ │ ├── stations.csv # 台站元数据站号、站名、省份、经纬度 │ └── weather/ # 按年拆分的日值观测文件都放这里 ├── scripts/ │ ├── etl.py # 读取、清洗、重采样 │ └── analysis.py # 趋势分析、极端事件统计、出图 └── output/ # 聚合结果和图片输出目录为什么把观测文件按年拆分而不是合成一个大CSV因为按年拆分后调试某个清洗规则时只需要对单年文件重跑能大幅缩短验证周期。等逻辑稳定了再用通配符一次性读全量。3. 气象数据预处理把几十万行观测读进Spark DataFrame3.1 数据长什么样站点、时间、要素、质量控制码从中国气象数据网下载的全国地面气象日值数据解压后通常是按年或按月拆分的多个CSV文件。每个文件里一行就是一个台站某一天的观测记录字段构成大致如下字段示例类型说明station_id54511字符串台站号5位数字可关联台站元数据date20130101字符串日期按YYYYMMDD格式存储tem12.5浮点日平均气温单位摄氏度pre0.0浮点日降水量单位毫米rhu43浮点相对湿度单位百分比qc0整型质量控制码0正常非0为可疑或错误实际字段名以你下载到的数据说明为准不同批次的数据集字段命名会有差异但结构基本逃不出“台站—时间—要素—质控”这套框架。如果数据源给的是JSON切片spark中读取json直接用spark.read.json(path)Spark会按JSON的嵌套结构自动推断schema不过气象数据还是CSV更常见后续示例都以CSV为主。3.2 读取与schema让Spark猜类型会翻车第一次读这类数据时最容易犯的错误是不指定schema直接让Spark推断。CSV里一旦出现缺测的哨兵值或空字符串Spark就会把这个本应是数值的列推断成字符串后面的avg(tem)要么直接报类型错误要么把字符串强转成null结果全废。我一般会先手工定义schema再读取from pyspark.sql import SparkSession from pyspark.sql.types import (StructType, StructField, StringType, DoubleType, IntegerType) spark SparkSession.builder \ .appName(weather-etl) \ .master(local[*]) \ .getOrCreate() # 手动定义schema防止Spark把数值列推断成string schema StructType([ StructField(station_id, StringType(), True), StructField(date, StringType(), True), # 日期先用字符串存入清洗时再转时间戳 StructField(tem, DoubleType(), True), StructField(pre, DoubleType(), True), StructField(rhu, DoubleType(), True), StructField(qc, IntegerType(), True), ]) df spark.read.csv( data/weather/*.csv, # 通配符读取该目录下所有CSV schemaschema, headerTrue, encodingUTF-8, # 如果中文站名乱码改成 GBK 再试 nullValue, ) df.printSchema() # 读完后先确认列类型符合预期 df.show(5, truncateFalse)这段逻辑的要点在于日期列暂时保留为字符串因为后面要用to_timestamp做格式转换提前转反而增加解析复杂度。nullValue告诉Spark把空字符串当作null处理而不是塞一个空值进去干扰统计。encoding参数是中文数据集的常见坑部分历史气象站点名字段是GBK编码用UTF-8读出来全是乱码逐个试一下编码是正常操作。提示读入任何数据集后先跑一次df.printSchema()再决定要不要继续往下做。这一步能发现80%的类型和编码问题。3.3 清洗哨兵值、物理范围、站点元数据join气象数据里的缺测值通常不是NaN而是“9999、32766”这类哨兵整数。直接用dropna根本筛不掉它们求均值时这些异常编码会被当成正常数值参与计算直接把温度拉高十几度。我第一次跑出来的全国图“南方比北方还热”就是因为这个。from pyspark.sql import functions as F # 把常见缺测哨兵值统一替换成 null后续聚合自动跳过 df_clean df.replace({ tem: {9999.0: None, 32766.0: None}, pre: {9999.0: None, 32766.0: None}, rhu: {9999.0: None, 32766.0: None}, }) # 物理范围校验中国地面站的常规气温不会超出 -50~50 度 df_clean df_clean.filter( (F.col(tem).isNull()) | ((F.col(tem) -50) (F.col(tem) 50)) ) # 降水量和湿度也要做范围约束 df_clean df_clean.filter( (F.col(pre).isNull()) | ((F.col(pre) 0) (F.col(pre) 600)) ) print(before:, df.count(), after:, df_clean.count())replace的键类型必须和列类型严格一致。schema里tem是DoubleType那字典键就要写9999.0而不是9999否则替换不生效且不报错这类静默失败最耗时间。温度范围过滤我保留了null行因为后面聚合时null会被跳过不影响统计但如果你要算的是“有效观测天数”就要先决定null是计入还是排除。站点元数据表通常只有几百行包含站号、站名、省份、经纬度、海拔。join这种小表时一定要用broadcast让Spark把小表复制到每个Executor节点避免触发shuffle这是气象项目里最常用也最有效的优化手段# 台站元数据表结构类似station_id, station_name, province, lat, lon, elevation station_schema StructType([ StructField(station_id, StringType(), True), StructField(station_name, StringType(), True), StructField(province, StringType(), True), StructField(lat, DoubleType(), True), StructField(lon, DoubleType(), True), StructField(elevation, DoubleType(), True), ]) stations spark.read.csv( data/stations.csv, schemastation_schema, headerTrue, encodingUTF-8, ) # broadcast 广播小表join 时不产生 shuffle df_clean df_clean.join(F.broadcast(stations), station_id, left)join类型选left是因为要保留全部观测记录如果某些站点的元数据缺失join之后province字段会是null这是后续按地区分析时需要排查的数据质量问题可以单独统计一下缺失比例。3.4 时间重采样字符串转时间戳并按年/月聚合日期列是“20130101”这种字符串不能直接提取年和月。用to_timestamp先转成时间戳再提取年份和月份进去这是整个流程里承上启下的一步from pyspark.sql import functions as F df_clean df_clean.withColumn( date_parsed, F.to_timestamp(date, yyyyMMdd) ) \ .withColumn(year, F.year(date_parsed)) \ .withColumn(month, F.month(date_parsed)) # 按年月重采样顺便统计每个月的有效温度记录数 monthly df_clean.groupBy(year, month).agg( F.avg(tem).alias(tem_mean), F.avg(pre).alias(pre_mean), F.count(tem).alias(tem_cnt), ) monthly.orderBy(year, month).show(12, truncateFalse)to_timestamp的格式串yyyyMMdd必须和原始字符串严格对应。如果数据里混了“2013-01-01”这种带横杠的格式解析结果就是null而且不报错。遇到大量null时先df.select(date).distinct().show()看看实际存在哪几种格式再做针对性处理。tem_cnt这个字段容易被忽略但它很重要有的月份站点上报率低平均值代表性差。后面做全国趋势分析时需要再次按台站数量过滤不能只看平均值就下结论。4. 核心分析指标从“读得动”到“算得出全国趋势”4.1 全国年均温趋势先看台站数量再信平均值全国尺度年均温最直接的计算方式是groupBy(year)对温度求平均。但这里有个数据质量陷阱不同年份的观测台站数量并不一致早期站点少、后期站点多直接平均会引入站点数量变化带来的偏差。比如某年新增的100个站点恰好分布在南方这一年全国均值就会偏高但这不是气候信号是样本结构变化。from pyspark.sql import functions as F trend df_clean.groupBy(year).agg( F.avg(tem).alias(annual_mean_tem), F.countDistinct(station_id).alias(station_cnt), F.count(tem).alias(obs_cnt), ) # 只保留台站数量足够多的年份避免样本结构偏差 trend_filtered trend.filter(F.col(station_cnt) 100) trend_filtered.orderBy(year).show(30, truncateFalse)countDistinct(station_id)统计的是参与计算的站点数量count(tem)统计的是温度有效记录数因为null会被跳过。我用station_cnt 100作为过滤条件意思是台站数量不足100个的年份不参与趋势分析这个门槛可以根据数据集的实际情况调整。如果数据集只有最近十年可以适当降低如果是五十年长序列100个站点的门槛能让趋势线更稳健。4.2 极端事件高温日、暴雨日的条件计数极端事件统计是气象分析里复用价值最高的部分。以高温日和暴雨日为例阈值遵循气象业务常用标准日最高气温达到35度计为高温日日降水量至少50毫米计为暴雨日。这种条件计数的写法比“先filter再count”高效得多from pyspark.sql import functions as F # 条件计数一次 groupBy 同时算高温日数、暴雨日数 extreme df_clean.groupBy(station_id, year).agg( F.sum( F.when(F.col(tem) 35, 1).otherwise(0) ).alias(hot_days), F.sum( F.when(F.col(pre) 50, 1).otherwise(0) ).alias(rainstorm_days), F.count(tem).alias(obs_days), ) extreme.filter(F.col(obs_days) 300) \ .orderBy(F.col(hot_days).desc()) \ .show(20, truncateFalse)这里用sum(when(...))实现条件计数而不是分别filter再count好处是只扫一遍数据就能同时算出多个指标。obs_days 300的过滤是为了剔除缺测严重、年有效观测天数不足的站点避免把“数据缺失严重”误判成“极端事件少”。如果你要统计的是极端低温或者大风日数只需要新增一个when分支其他逻辑完全不变。spark内存问题最容易出现在这一步。原因在于groupBy(station_id)会把同一个站点的所有记录哈希到同一个分区如果某个站点文件异常巨大单个分区的数据量远超其他分区堆内存就容易被撑爆。遇到这种情况先看日志里shuffle阶段的报错再考虑增加分区数或对站点列做加盐处理。4.3 区域对比与可视化结果小而精时再回Pandas有了省份信息之后区域维度比全国均值更有分析价值。按省份聚合全国两三千个站点的数据结果集只有几十行这时把结果拉回Python生态做可视化是完全安全的from pyspark.sql import functions as F province_yearly df_clean.groupBy(province, year).agg( F.avg(tem).alias(tem_mean), F.countDistinct(station_id).alias(station_cnt), ) # 先过滤再聚合结果集已经很小 province_means province_yearly \ .filter(F.col(station_cnt) 20) \ .groupBy(province) \ .agg(F.avg(tem_mean).alias(province_longterm_mean)) \ .orderBy(F.col(province_longterm_mean).desc()) province_means.show(35, truncateFalse)这里需要强调一个边界原始观测数据绝对不能直接toPandas()。几十万行数据拉回Driver进程内存直接爆掉是轻的严重时整个IDE会卡死。只有像上面这样经过聚合、结果集缩小到几十行时toPandas()才是安全的。# 聚合后的结果集已经很小collect 到本地再用 matplotlib 画图 pdf trend_filtered.orderBy(year).toPandas() import matplotlib.pyplot as plt plt.figure(figsize(10, 5)) plt.plot(pdf[year], pdf[annual_mean_tem], markero, linestyle-) plt.title(China Annual Mean Temperature (Weather Stations)) plt.xlabel(year) plt.ylabel(temperature (degree C)) plt.grid(True) plt.savefig(output/china_annual_mean_temp.png, dpi150) print(saved to output/china_annual_mean_temp.png)这是python数据分析与可视化实践里最常见的组合Spark负责算Pandas和Matplotlib负责呈现。如果结果集更大一点需要保留全量明细时就用df.write.parquet(output/xxx.parquet)落盘Parquet格式既保留schema又压缩体积后续想用Spark再读回来也方便。5. 避坑Spark做气象分析最常见的5个坑这些坑都是我在跑类似数据时真实踩过的按“现象、原因、解决”三条写清楚每一条都能对应到前面某一个环节。5.1 数据侧的3个坑哨兵值、类型推断、日期解析坑1年均温整体偏高画出来的趋势图离谱。现象avg(tem)算出来的全国年均温比气候公报高好几度曲线形状完全不符合常识。原因缺测哨兵值9999、32766这类编码没有被替换成null直接被当成正常数值参与了平均计算。气象数据的缺测不是NaN不能用dropna处理。解决读取后立即做哨兵值映射把已知的缺测编码统一替换成null再做聚合统计。替换时注意键类型必须和列类型一致DoubleType列要用9999.0不能用9999。坑2读CSV后列名全是_c0、_c1数值列变成string。现象df.printSchema()显示所有列都是string列名是_c0、_c1这种默认名字后续avg()直接报类型错误。原因文件第一行不是标准表头或者Spark在读取时没有识别到header又因为文件中混入了非数值内容导致类型推断失败。解决读取时显式指定headerTrue并手动定义schema不要依赖Spark自动推断。读完后立刻printSchema()检查确认数值列是DoubleType再继续。坑3to_timestamp之后日期列全是null。现象按月聚合结果里年份全是null统计直接报废。原因日期字符串里混用了“20130101”和“2013-01-01”等多种格式单一格式串无法覆盖全部数据解析失败静默返回null。解决先用df.select(date).distinct().show()检查实际格式种类再针对性处理。如果格式确实混杂可以用unix_timestamp尝试多个格式串或者写when分支分步解析。5.2 执行侧的2个坑collect炸内存、文件级数据倾斜坑4collect()或toPandas()直接把Driver内存打爆。现象跑分析时Driver端报OutOfMemoryErrorIDE卡死甚至需要强制重启。原因对未聚合的原始DataFrame直接调用collect()几十万行数据一次性拉回Driver进程内存瞬间耗尽。解决先做聚合、过滤确认结果集只有几十到几百行时再collect()或toPandas()。需要全量明细时改用df.write.parquet(output/xxx)落盘不要拉回本地。坑5某个分区的数据量远超其他分区单个Executor OOM。现象日志里shuffle阶段报内存溢出或者某个Stage长时间卡住不动重跑几次结果一样。原因groupBy(station_id)时某些台站的观测记录特别多比如大城市站点或合并文件默认哈希分区把它们打到了同一个分区数据倾斜导致单点压力过大。解决先看键的分布情况统计每个站点的记录数确认倾斜程度然后调整分区数或对站点列加盐把key拆成多个子key让热点数据分散到不同分区。温度、降水这类分析基本是读多写少确认中间结果被多个分析复用时可以cache()用完记得unpersist()。6. 进阶三个调参习惯和一条结果验证路径6.1 调参习惯shuffle分区、executor内存、缓存复用三个参数是Spark跑这类数据项目的关键但它们的生效条件完全不同别混为一谈。spark.sql.shuffle.partitions控制shuffle后生成的分区数默认是200。做全国站点月聚合这种数据量200个分区会产生大量小文件写入低效。我一般会调低到20左右让数据落盘更集中。但如果你在跑大表join分区数太低又会拖慢并行度需要根据数据量来回试。spark.executor.memory在本地模式没有意义因为Driver和Executor在同一个JVM进程里你设置了也不生效。它只在通过spark-submit --master yarn提交集群时才有真正作用本地开发阶段不用纠结这个参数。中间结果复用是性价比最高的优化。清洗后的DataFrame如果被趋势分析、极端事件统计、区域对比多个环节使用就值得cache()一下让数据物化在内存里避免每个分析都重新读一遍源文件。但缓存不是免费的用完了记得unpersist()释放内存否则缓存垃圾会越积越多后续任务反而变慢。6.2 阈值参数外置让高温暴雨定义可改气象分析里“高温”“暴雨”的阈值定义会随着业务需要调整。别把35度和50毫米硬编码在代码里放到配置文件中import json with open(config.json, r) as f: cfg json.load(f) hot_threshold cfg[thresholds][hot_temp] # 默认 35.0 rainstorm_threshold cfg[thresholds][rain] # 默认 50.0 missing_vals cfg[missing_values] # [9999.0, 32766.0]这样改阈值只需要动配置文件不用碰代码。极端事件统计、哨兵值替换、物理范围过滤都可以参数化整个项目的可维护性会明显提升。6.3 结果验证拿一个站点对照官方气候值流程跑通后别急着相信图表先做一条验证路径单独取出北京站站号54511近十年的逐年温度和国家气候公报里北京站的公开值对比误差在0.5度以内说明整个清洗和聚合链路是可信的如果偏差大回头看哨兵值是否清理干净、schema是否把数值列推断成了string。趋势图也要对照公开的气候趋势描述看看整体形状是否吻合。我自己的习惯是任何数据分析项目都先做一个“已知答案的单点验证”确认链路可信后再放大到全国尺度。如果单点都对不上全量结果再漂亮也不能信。我就是这么栽过一回的第一次跑全国气温项目画出来的趋势图南方明显比北方热查了半天才发现32766这个哨兵值被当成有效温度参与了平均。后来养成了先做数据完整性检查、再跑分析的规矩这类翻车就基本消失了。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
