基于Spark的气象数据处理与分析：从爬虫到可视化完整实战-北京尧图网络

简介面向大数据期末课设的Spark气象数据处理与分析完整项目报告适合大数据专业学生及Spark入门开发者参考。报告以中央气象台全国2412个城市、57888条24小时气象数据为对象完整呈现了从省份、城市及天气数据逐层爬取到Ubuntu环境下Spark与Jupyter Notebook配置、DataFrame清洗转换、缺失异常值处理、groupBy城市聚合计算平均气温与降水量再到matplotlib可视化展示的整套实验流程。资源共1个文件为DOCX格式课程设计报告大小744KB内含项目背景、实验环境、数据获取方案、核心算法与代码、结果图表与总结等模块步骤清晰、代码注释明确对缺失值与异常值处理、城市分组聚合等关键环节给出了可复现的实现思路。已有13405人学习下载对正在完成Spark课设选题或希望掌握气象数据分析完整链路的学习者具有实际帮助。1. 这门课设要解决的不是「预测天气」而是「让气象数据变得可算」天气预报每天都在用但真正做大数据课设时你会发现气象数据本身并不那么「干净」。以我经手的这个基于 Spark 的气象数据处理与分析课设为例它的目标不是训练预测模型而是完成一条完整链路从公开气象接口爬取全国城市的整点观测数据用 Spark 做清洗和聚合算出「过去 24 小时累计雨量」和「当日平均气温」最后用图表把结果呈现出来。对大数据专业的期末课设来说这套流程恰好覆盖了数据采集、数据清洗、分布式计算和数据可视化四个核心环节也对应了Spark 实战中最常被考察的 DataFrame 操作能力。这个选题适合两类人一是正在准备大数据期末课设、需要一份能讲清楚「为什么这么设计、每一步在做什么」的完整案例的学生二是想快速上手 Spark 处理真实 CSV 数据的从业者——这里的数据可不是教材里那种规规矩矩的示例表有缺失、有异常、有时次不齐处理起来才有意思。下面我按「实验环境 → 数据获取 → 分析建模 → 避坑 → 验收答辩」的顺序把整套方案摊开来讲。2. 总体架构设计三层数据链路与 Spark 分析模块的职责划分2.1 实验环境版本搭配与可视化组件安装我当时的环境是 Ubuntu 16.04 Python 3.9 Spark 2.4.0 Jupyter Notebook。这套组合在 2022 年属于比较经典的课设配置Spark 2.4.0 对 DataFrame API 的支持稳定Python 3.9 能兼容绝大多数数据分析库Jupyter 则方便分步调试和展示中间结果。如果用的是更新的 Spark 3.xAPI 上的差别不大但要注意 Scala 和 Java 的版本匹配否则启动时会报错。环境装好后还需要补两个可视化依赖命令如下sudo apt-get install python3-matplotlib sudo apt-get install python3-tkmatplotlib 负责绘图python3-tk 是 Tkinter 的底层支持没有它很多绘图后端会报「No module named _tkinter」。这里要提醒一句如果你是在虚拟环境里装 Python不要只 pip install 了事系统层面的 apt 依赖缺失是这类环境问题里最容易翻车的点而且报错信息往往不直接——通常会在你画图那一步才蹦出来。2.2 数据链路设计三级接口逐层递进要理解整个系统的数据流可以先看这张图数据从气象台网站的异步接口来经过三层请求才能到达分析模块。省份列表接口province ↓ 获取省份编码城市列表接口province/{省份编码} ↓ 获取城市编号 24 小时整点天气接口passed/{城市编号} ↓ 解析 JSON、补充字段 CSV 落盘province.csv / city.csv / passed_weather_ALL.csv ↓ Spark SQL 读取分析结果雨量排序、日均温排序 ↓ collect / head matplotlib 可视化这个设计的关键在于网页上的数据是 JSON 格式异步加载的省去了解析 HTML 的麻烦。每层的 URL 有明确规律——省份数据、某个省份的城市数据、某个城市最近 24 小时整点数据——所以爬虫只需要依次请求、解析、存储。完整数据规模是 2412 个城市、57888 条记录每条记录包含整点时间、气温、降水量、风力、气压、相对湿度等字段。2.3 数据结构设计CSV 字段含义与存储约定为了方便后续 Spark 读取和异常辨识爬取后的数据统一保存为 CSV 格式核心字段如下字段名含义类型备注province城市所在省份中文文本用于分组city_name城市名称中文文本用于展示city_code城市编号数字字符串作为城市唯一标识city_index城市序号计数数字爬取顺序编号time时间点整点字符串含日期与小时temperature整点气温数值需 cast 为数值类型rain1h过去 1 小时降雨量数值可能含异常值pressure整点气压数值本实验未使用humidity相对湿度数值本实验未使用我一般会把input和output两个目录分开爬虫写inputSpark 分析结果写output这样重跑分析时不用重新爬数据也方便检查中间产物。接下来就到了数据获取模块的具体实现。3. 数据获取模块JSON 接口分析、断点续爬与 CSV 落盘3.1 从浏览器控制台定位接口规律拿到一个数据源第一步不是写代码而是观察。打开气象台网站任意点一个热点城市在 Chrome 的 Web 控制台里切换省份和城市就能看到网络请求的返回体是 JSON 字符串。通过对比请求 URL 和返回数据可以归纳出三个固定模式省份数据走/f/rest/province某个省份的城市数据走/f/rest/province/拼接省份三位编码最近 24 小时整点天气数据走/f/rest/passed/拼接城市编号。每个城市的编号比如福州是 58847要从城市接口的返回里取城市接口的数据又依赖省份接口里的编码。所以爬取顺序是固定的先省份、再城市、最后逐城市抓天气数据不能反过来。这个「接口依赖链条」是整个爬虫模块的核心设计理清了它后面代码就是体力活。3.2 爬虫核心实现请求、解析、落盘三段式我用 Python 实现了一个 Crawler 类把请求、解析、存储拆成独立方法。核心逻辑如下import urllib3 import json import csv import codecs import os import time class Crawler: def get_html(self, url): # 发送 HTTP 请求并读取响应体 request urllib3.Request(url) response urllib3.urlopen(request) return response.read() def parse_json(self, url): # 响应体直接就是 JSON 字符串无需解析 HTML obj self.get_html(url) json_obj json.loads(obj) if obj else list() return json_obj def write_header(self, file, data): # 写入 CSV 表头注意加 BOM 防止 Excel 打开中文乱码 if data: with open(file, ab) as f: f.write(codecs.BOM_UTF8) f_csv csv.DictWriter(f, data[0].keys()) f_csv.writeheader() f_csv.writerows(data) def write_row(self, file, data): # 追加写入数据行不重复写表头 if data: with open(file, ab) as f: f_csv csv.DictWriter(f, data[0].keys()) f_csv.writerows(data)这个设计里有一个细节值得注意write_header和write_row分开。第一次落盘时写表头后续每攒够 50 个城市只追加数据行。如果你每次都调writeheaderCSV 里会出现几十行重复表头后面 Spark 用spark.read.csv读取时会把表头当数据直接导致字段类型推断错乱。这也是我在课设里踩过的第一个坑。3.3 按省份递归获取城市按 50 城市批量落盘省份和城市数据的获取逻辑类似文件不存在才爬存在就直接读本地天然支持断点续爬。天气数据量大每抓完一个城市都写盘太慢、内存也扛不住所以我用count % 50 0作为批量落盘的触发条件def get_passed_weather(self, province): weather_passed_file input/passed_weather_ province .csv if os.path.exists(weather_passed_file): return # 已有数据则跳过支持断点续爬 passed_weather list() count 0 if province ALL: for city in self.get_cities(): data self.parse_json(http://www.nmc.cn/f/rest/passed/ city[code]) if data: count 1 for item in data: item[city_code] city[code] item[province] city[province] item[city_name] city[city] item[city_index] str(count) passed_weather.extend(data) time.sleep(1) # 频率控制避免请求过快 if count % 50 0: if count 50: self.write_header(weather_passed_file, passed_weather) else: self.write_row(weather_passed_file, passed_weather) passed_weather list() # 清空内存缓存time.sleep(1)不是可有可无的装饰是防止请求频率过高被服务端限流。全量爬完 2412 个城市需要半小时以上所以正常流程是爬一次之后所有分析都直接读input/passed_weather_ALL.csv。如果哪次数据爬坏了删掉对应 CSV 重跑即可变量名、文件名都指向清楚的输入输出约定整个链路不太需要额外维护逻辑。3.4 数据校验为什么「有数据」不代表「数据可用」爬虫返回的是 JSON 列表但有些城市可能不存在或者 24 小时数据全部缺失。parse_json里json_obj or list()的写法就是处理这种情况请求成功但没有内容时返回空列表后续if data判断会跳过这个城市不会把空数据写进 CSV。这也是数据清洗流程的第一道关卡。好消息是这个气象站点的接口返回格式非常规范省去了 HTML 标签解析和字符集猜测的麻烦。坏消息是城市列表里有部分城市编号在天气接口下返回空还有一些整点的数值明显异常比如降雨量上千。这些坑我会在第 5 章统一展开先继续看 Spark 分析部分。4. 分析模块实战Spark SQL 计算累计雨量与四时次日均温4.1 创建 SparkSession 并加载 CSV分析阶段用 Spark SQL 的 DataFrame API 完成。第一步创建 SparkSession 并读取爬取结果from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WeatherAnalysis) \ .config(spark.some.option, some-value) \ .getOrCreate() df spark.read.csv(input/passed_weather_ALL.csv, headerTrue, inferSchemaTrue, encodingutf-8)inferSchemaTrue会让 Spark 自动推断字段类型这样temperature和rain1h读进来就直接是数值类型不需要在 SQL 里反复CAST。需要留意的是 CSV 里的 BOM 头Spark 读取时要确认 encoding 参数否则第一列列名可能带一个肉眼看不见的\ufeff前缀导致后续select字段名匹配失败——这也是我实际踩过的坑之一。4.2 核心计算一过去 24 小时累计雨量计算思路很直接按城市分组对rain1h求和。但直接求和会在异常数据上翻车所以要先过滤再聚合from pyspark.sql import functions as F # 选择字段并将 rain1h 显式转为数值类型 df_rain df.select( F.col(province), F.col(city_name), F.col(city_code), F.col(rain1h).cast(double) ) # 过滤异常降雨量记录大于 1000 视为脏数据 df_rain df_rain.filter(F.col(rain1h) 1000) # 按城市分组求和得到过去 24 小时累计雨量 df_rain_sum df_rain.groupBy(province, city_name, city_code) \ .agg(F.sum(rain1h).alias(rain24h)) \ .sort(F.col(rain24h).desc()) # 缓存中间结果避免重复计算 df_rain_sum.cache() # 合并分区为一写回本地 CSV df_rain_sum.coalesce(1).write.csv(output/rain24h_summary) # 取出前 20 的城市列表供可视化展示 top20_rain df_rain_sum.head(20)这里每个步骤都有明确的目的先cast是因为 CSV 可能把数值读成了字符串再filter是为了剔除降雨量大于 1000 的异常记录cache()缓存了前面一系列转换的中间结果之后head()和write复用同一份计算不会重复跑coalesce(1)把结果收敛到单分区避免写出几十个小文件——这个操作在展示成果时非常重要老师点开你的 output 文件夹看到一堆 part 文件会很头疼。4.3 核心计算二当日平均气温严格按四时次标准日平均气温比雨量麻烦一点。国家标准《地面气象服务观测规范》要求取 02 时、08 时、14 时、20 时四个时次求平均不能简单地对所有整点取均值。实现上要经历「拆时间 → 筛时次 → 分组计数 → 过滤不完整样本」四个阶段from pyspark.sql.functions import hour, date_format # 1. 选择气温字段并将 time 拆分出日期和小时 df_temperature df.select( F.col(province), F.col(city_name), F.col(city_code), F.col(temperature).cast(double), date_format(F.col(time), yyyy-MM-dd).alias(date), hour(F.col(time)).alias(hour) ) # 2. 只保留 02/08/14/20 四个时次的记录 df_4point_temperature df_temperature.filter( F.col(hour).isin(2, 8, 14, 20) ) # 3. 按城市日期分组求和并计数 df_avg_temperature df_4point_temperature.groupBy( province, city_name, city_code, date ).agg( F.sum(temperature).alias(sum_temp), F.count(temperature).alias(cnt) ) # 4. 过滤掉不足四个时次的记录确保日均温计算完整 df_avg_temperature df_avg_temperature.filter(F.col(cnt) 4) \ .withColumn(avg_temperature, F.format_number(F.col(sum_temp) / 4.0, 1)) \ .select(province, city_name, city_code, date, avg_temperature) \ .sort(F.col(avg_temperature).desc()) df_avg_temperature.cache() df_avg_temperature.coalesce(1).write.json(output/avg_temperature) # 4. 收集到 Driver 端方便后续 matplotlib 可视化 result_list df_avg_temperature.collect()这里最关键的设计是用cnt 4做完整性校验。因为爬取过程有一个实际约束部分城市的 02 时数据可能缺失这时对 3 个时次求平均得到的结果没有参照价值甚至会低于真实值。加了计数过滤后日均温的结果才和气象台公布的数字对得上——这个细节在答辩时讲出来是加分项。4.4 为什么用 DataFrame API 而不是 RDD有同学问过这个场景用 RDD 也能做为什么非用 DataFrame两个原因。第一DataFrame 的groupBy().agg()比 RDD 的reduceByKey简洁得多代码量大概少一半第二DataFrame 有 Catalyst 优化器filter下推和cache之后的重复查询利用能做到比较高效Spark 2.4 里这些优化是自动的不需要手写。课设的价值不只是能跑出结果更重要的是让评审看到你「知道有几种写法、为什么选这种」这才是本科阶段该有的工程判断力。分析结果出来后两项计算结果各生成一份排序文件rain24h_summary.csv给 Top 城市排名avg_temperature.json给日均温列表。两者都coalesce(1)收敛到单文件然后head()或collect()取出后交给 matplotlib 绘图。5. 避坑指南数据缺失、时次不全、中文乱码与分区爆炸5.1 序列化问题在代码里表现为「过滤无效」现象明明 CSV 里的rain1h看起来是数字但filter(rain1h 1000)过滤后结果数量不对某些明显异常的记录仍留在结果里。原因CSV 读取时inferSchema未能正确推断数值类型字段被读成了字符串「1000」和 1000 的比较变成字符串比较。解决在聚合之前显式cast(double)并在写完分析结果后打印一行df.printSchema()确认字段类型。这是我在课设里遇到的第一类「玄学问题」后来变成习惯凡是参与算术运算的字段一律先 cast 再操作绝不在字符串上赌类型推断。5.2 日均温偏低四时次数据不齐的隐蔽陷阱现象某城市日均温排序结果明显偏低和气象台公布值差 2 度以上。原因该城市 02 时或 20 时数据缺失实际只有 2 到 3 个时次参与平均。尤其是 20 时数据在当天 20 时 30 分前爬取时必定缺失导致平均温度被「夜间低温」和「清晨低温」拖低。解决groupBy后count()再用filter(cnt 4)剔除不完整样本。另一个辅助手段是尽量在 20 时 30 分之后启动爬虫保证当天四时次数据齐全。这也是为什么项目里反复强调「建议在当天 20 时 30 分后再爬取数据」——天气数据接口的时次更新不是实时的过早爬取拿不到 20 时整点记录日平均气温计算就先天残缺。5.3 CSV 打开乱码与表头重复现象Excel 打开input/*.csv中文变成乱码或者文件里出现多行表头。原因缺少 UTF-8 BOM 头writeheader被重复调用。解决写入前加codecs.BOM_UTF_8函数设计上把「写表头」和「追加写行」分开只在count 50时执行一次write_header后续统一write_row。这个设计在 3.2 节已经给出代码。有人觉得 BOM 是老古董但在国内用 Excel 看 CSV 的场景下有没有 BOM 完全是两个体验。多写一行f.write(codecs.BOM_UTF8)能省去后续所有「字符集猜谜」这笔账怎么算都划算。5.4 输出目录里几十个小文件现象output/rain24h_summary/目录下不是干净的一个 CSV而是几十个part-000XX文件。原因Spark 默认按集群分区数输出即使课设是本地模式分区数也可能大于 1。解决coalesce(1)合并分区再写入。同理df_avg_temperature.write.json之前也做同样的收敛。这个问题的现实影响是老师打开你的成果目录看到一堆part-文件会质疑你没有控制输出或者直接问你「哪一个是最终结果」。coalesce(1)一行代码就能避免这种无谓的扣分。5.5 可视化阶段才发现的缺失字段现象matplotlib 画图时发现某些城市没有数据图上出现空洞。原因df_avg_temperature.filter(cnt 4)在剔除不完整样本的同时把部分城市整体过滤掉了。解决可视化时先用result_list里的城市列表核对原始 CSV确认是该城市数据真的缺失而不是代码 bug。处理方式有两种结果图上标注「该城市数据缺失」或者退一步改为「三点平均」并备注方法差异。「过滤掉」不是唯一出路关键是你得能解释为什么这些城市不在结果里。6. 验收与答辩怎么把课设从「跑通了」讲到「设计对了」6.1 验证链路是否完整按照这个清单逐项验收验收项操作方法通过标准数据量核对读取passed_weather_ALL.csv行数约 57888 条允许部分缺失省份数据完整对比province.csv条数与接口返回无缺漏省份雨量异常过滤打印df_rain过滤前后计数过滤后无 1000 记录日均温合理性随机抽 5 个城市与气象台公开值对比差值在 1 度以内输出文件格式查看 output 目录每个结果一个文件无 part 散件可视化图表确认图表标题、坐标轴标签完整无中文乱码、无空图答辩时我一般会现场演示一个「数据抽查」随机选一个城市从原始 CSV 里筛出它的四个时次数据手算一遍平均气温再和avg_temperature.json里的值做对比。这一步能让评委直观看到你的计算链路没有「黑匣子」——Spark 算出来的每个数都能回溯到原始输入。6.2 答辩技巧准备好三个方向的追问评委大概率会问三类问题第一数据为什么有缺失缺失了怎么处理第二为什么日平均气温只取四个时次第三cache()和coalesce(1)分别解决了什么问题前两个问题在 3.4 和 4.3 节已经答过核心是把「国家标准规定」和「计算完整性校验」结合起来说。第三个问题可以展开讲一遍血缘关系cache保存的是 DataFrame 经过转换后的中间结果coalesce改变的是物理分区数两者一个优化重算、一个收敛输出。还有一个小技巧把第 5.1 节的类型推断问题作为「踩坑收获」主动讲出来。与其等评委发现不如在展示时直接说「这里我遇到过一个字符串比较的坑后来通过 cast 解决」这样既展示了排查能力也把面试官可能追问的方向提前堵住。答辩本质上不是展示完美而是展示「怎么发现问题、怎么定位问题、怎么低成本解决问题」的完整思路。6.3 一个关于「演示顺序」的教训我印象很深的一次答辩有个同学演示时直接从爬虫代码开始讲结果前面讲了 5 分钟接口格式后面 Spark 核心计算只剩 3 分钟评价自然不理想。从那以后我每次做这类演示都会强制走一遍「结果先行」的顺序先把可视化结果摆出来让评委看到「计算出了什么」再倒推「爬虫怎么设计、Spark 怎么算」。这样既保证核心内容有充足时间也让评审带着结果去理解过程。希望这套方案能帮你在课设答辩前把链路理得更顺——数据能算、结果能查、踩坑能讲这三点都做到了课设自然就立住了。本文还有配套的精品资源点击获取

基于Spark的气象数据处理与分析：从爬虫到可视化完整实战

相关推荐

直播美颜SDK性能优化全解析：架构、算法与踩坑实录

RAG检索优化实战：分块、混合召回与重排核心方案

大模型Agent开发全流程：模型选型、工具设计、护栏与编排

更多资讯

YOLOv6工业视觉检测优化与部署实战：从能检出到敢上线

Seedance 2.5 视频生成提示词完全指南：从底层逻辑到实操避坑

STM32寄存器白话手册：从地址映射到位操作的系统级理解

Ollama本地大模型部署全流程：硬件、量化、调优与接口实战

WSL烧录STM32实战：用GDB Server桥接绕过USB限制

从照亮到编码：光度立体、相移与格雷码的逻辑与工程实现

今日更新

Verilog实现FPGA密码锁：矩阵键盘与状态机设计实战

数据服务监控运维实战：从指标体系到告警治理与故障排查

FreeRTOS任务通知实战详解：内核原理、API与避坑指南

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？