Spark气象数据处理实战：从爬虫到Top-N分析的全流程

简介一套面向大数据期末课设的Spark气象数据处理与分析课程设计报告完整覆盖从项目背景、实验环境搭建到数据获取、清洗、聚合分析与可视化的全流程能帮助读者快速掌握Spark在大数据场景下的实际应用。实验基于Ubuntu 16.04、Python 3.9与Spark 2.4.0爬取中央气象台2412个城市24小时共57888条气象数据重点讲解各省份、城市及天气数据的JSON接口解析与三级爬取策略并演示用DataFrame处理缺失值与异常值、按城市分组计算平均气温与降水量的核心操作还可利用窗口函数分析时间序列变化趋势配合matplotlib展示气温变化、降水量分布等可视化图表。资源包共1个docx文档压缩包大小744KB文档内容详实涵盖完整实验步骤与代码适合大数据专业学生作为期末课设参考也可用于Spark实战入门。目前已有13400人学习下载报告包含完整目录、项目背景、实验环境、爬虫源代码及分析代码逻辑清晰、注释详细可直接借鉴其代码框架与实验流程按步骤复现气象数据处理全链路。1. 气象数据处理与分析一份能直接复现的 Spark 课设做大数据期末课设时与其搭一个假数据演示不如把真实业务链路跑通从中央气象台拉取全国 2412 个城市的整点天气数据交给 Spark 做清洗、聚合、排序最后落盘成可直接可视化的 CSV 和 JSON。这套基于 Spark 的气象数据处理与分析方案覆盖了爬虫、Spark SQL、结果校验三个环节数据规模 57888 条能真实触发数据倾斜、缺失值、异常值这些课堂上讲不清的坑。适合正在做大数据课设、想要一份能跑出 Top-N 排行榜和日均温排序结果的实战案例的从业者也适合拿来替换自己课设里「加载 local 文件、show 一下」那种凑数流程。2. 先解决数据从哪来三个 JSON 接口与断点续爬2.1 目标站点的接口规律三层 URL 逐级下钻整个项目不需要解析 HTML气象站点把数据以 JSON 异步加载。某城市 A 用 Chrome 打开控制台切换省份时定位到三个请求 URL规律如下http://www.nmc.cn/f/rest/province # 返回省级列表 http://www.nmc.cn/f/rest/province/{省份编码} # 返回该省城市列表 http://www.nmc.cn/f/rest/passed/{城市编号} # 返回该城市最近24小时整点数据省份编码不是数字而是三位字母比如 A 省是“ABJ”城市编号则是类似“58847”的五位站号。三级数据存在父子依赖需要先拉省份列表再逐省拉城市最后按城市编号逐个请求天气数据。这种链路在真实数据采集里非常典型——搜索引擎、电商分类页、天气 App 的接口基本都是这个结构学会了这套「先拿索引再拿明细」的爬法换个目标站点也能复用。代码按顺序执行三层请求import json import os import csv import codecs import time import urllib.request class Crawler: def get_json(self, url): req urllib.request.Request(url) resp urllib.request.urlopen(req, timeout10) return json.loads(resp.read().decode(utf-8)) def get_provinces(self): province_file input/province.csv if not os.path.exists(province_file): provinces self.get_json(http://www.nmc.cn/f/rest/province) self.write_rows(province_file, provinces, write_headerTrue) return provinces return self.read_csv(province_file) def get_cities(self): city_file input/city.csv if not os.path.exists(city_file): cities [] for prov in self.get_provinces(): code prov[url].split(/)[-1].split(.)[0] cities.extend(self.get_json(http://www.nmc.cn/f/rest/province/ code)) time.sleep(0.5) self.write_rows(city_file, cities, write_headerTrue) return cities return self.read_csv(city_file)url.split(/)[-1].split(.)[0]这行是把省份详情 URL 里的编码抠出来气象站点的 URL 结构比较规整所以直接字符串切分即可。get_provinces和get_cities都做了「文件存在就跳过」的判断这是整个爬虫能断点续爬的关键——数据量一大中途断了不可能从头再来。A 同学的原始脚本里网传的 urllib2 是 Python 2 写法实际跑在 Python 3.9 下需要换成urllib.request上面代码已做兼容处理这是第一个易踩的版本坑。2.2 城市 24 小时数据与增量落盘拿到城市列表后进入核心的天气数据抓取。原始站点每个城市返回约 24 条整点记录字段包含时间、气温、降水量、风力、气压、湿度等。A 同学用「每 50 个城市落盘一次」来控制内存占用这是很务实的做法——全国 2412 个城市全部堆在内存里再统一写盘笔记本大概率直接卡死。def crawl_passed_weather(self, province_filterALL): target input/passed_weather_ALL.csv if os.path.exists(target): return passed [] count 0 for city in self.get_cities(): if province_filter ! ALL and city[province] ! province_filter: continue data self.get_json(http://www.nmc.cn/f/rest/passed/ city[code]) if data: # 过滤无数据城市 for item in data: item[province] city[province] item[city_name] city[city] item[city_code] city[code] passed.extend(data) count 1 time.sleep(1) # 控制请求频率 if count % 50 0 and passed: self.write_rows(target, passed, write_header(count 50)) passed [] if passed: self.write_rows(target, passed, write_headerFalse)city[code]必须补进每条天气记录否则后面按城市分组时找不到归属。write_header(count 50)的含义是只有第一批数据写入时才写表头后续追加写行这是避免 CSV 表头重复出现的常规做法。time.sleep(1)是对目标站点的基本礼貌不加这个 Sleep 爬完 2412 个城市大概率触发对方限流实测会连续返回空列表爬虫程序拿到空数据后 count 不再增长造成「假死」。def write_rows(self, file, rows, write_headerTrue): if not rows: return with open(file, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesrows[0].keys()) if write_header: writer.writeheader() writer.writerows(rows)写入时用utf-8-sig编码等于在文件头加了 BOMExcel 直接打开 CSV 就不会出现中文乱码。后期用spark.read.csv读取时注意加headerTrue否则第一行城市名会被当成数据。整段程序是标准的「索引表 明细表」增量抓取模型24 小时内跑一次能落地约 5.8 万条记录耗时约半小时到一小时视网络状况浮动。3. 用 Spark 算降雨量 Top-NDataFrame 分组聚合的完整链路3.1 数据加载与清洗read.csv 自动推断 Schema原始 CSV 里拿到的字段至少有 province、city_name、city_code、time、temperature、rain1h其中rain1h是过去 1 小时降雨量单位毫米气温单位摄氏度。第一步是用 SparkSession 读取本地 CSVfrom pyspark.sql import SparkSession from pyspark.sql.functions import col, cast spark SparkSession.builder \ .appName(weather-analysis) \ .master(local[4]) \ .getOrCreate() df spark.read.csv(input/passed_weather_ALL.csv, headerTrue, inferSchemaTrue, encodingUTF-8)master(local[4])代表本地用 4 个线程模拟分布式执行对于 5.8 万条数据足够用内存按 2G 之内就能扛住如果改了更大数据集需要调spark.driver.memory和spark.executor.memory。inferSchemaTrue会自动把 temperature 推断成 double、city_code 推断成 int但这个推断不是百分百可靠部分城市的字段缺失会导致整列被推断成 string所以后面聚合前仍要显式 cast。df.printSchema() df.groupBy(province, city_name).count().orderBy(col(count).desc()).show(5)验证数据规模时可以先做分组计数正常每个城市的 24 小时记录数是 24 条左右出现个别城市是 18 条或 30 条也不必惊讶——站点上线时间、网络重试都会造成少量缺失。3.2 降雨量累加三列分组 异常值过滤计算「过去 24 小时累计降雨量」是分组聚合的典型场景。A 同学的思路是按省份、城市名、城市编号三列分组对rain1h求和然后降序排序取 Top。这里城市编号必须存在因为不同省份可能有同名城市只按城市名分组会把两个同名城市的数据混在一起。df_clean df.filter(col(rain1h).isNotNull()) \ .filter(col(rain1h) 1000) df_rain_sum df_clean.groupBy(province, city_name, city_code) \ .agg(cast(col(rain1h), double) \ .cast(decimal(10,2))异常值过滤里rain1h 1000是一个比较粗暴但有效的规则降水量读取异常时会出现明显离谱的数字比如 9999 或暴雨站点的传感器故障值。这些数值必须在一开始就过滤掉否则一个异常值就能把城市降雨总量拉高几百倍直接污染 Top-N 排名。另一个工程习惯是保留过滤规则到最终落盘不要在 show 时只显示前 20 行就当分析完成。3.3 排序与落盘coalesce 让结果可预测计算完聚合后需要把结果降到单分区再写出。这一步有讲究df_rain_top df_rain_sum.orderBy(col(rain24h).desc()) df_rain_top.cache() df_rain_top.coalesce(1).write.csv(output/rain_top, headerTrue) top_list df_rain_top.head(30).toJson()cache()的作用是缓存中间结果后面的 head 和 write 都复用它避免重复计算整个血缘。.coalesce(1)把结果合并成单分区这是写小结果集的标准做法否则 Spark 默认会生成 200 个分区产出几十个空文件后续读取很不方便。A 同学的项目里很大一个坑就是忘了 coalesce我在某图像处理 Demo 里也踩过同样的坑——小结果集直接 write 会产生 200 个 part 文件必须手动合并。3.4 平均气温计算的时次过滤只有 4 个时次能算日均温气温部分比降雨量略复杂。国标日平均气温取 02 时、08 时、14 时、20 时四个时次数据的平均值因此不能直接对一天内所有整点求平均需要先过滤时次。A 同学用hour()函数抽取时间字段的小时部分df_temperature df.select( province, city_name, city_code, temperature, split(col(time), )[0].alias(date), hour(col(time)).alias(hour) ) df_four_points df_temperature.filter( col(hour).isin([2, 8, 14, 20]) )split(col(time), )[0]是把 2022-12-11 14:00:00 拆出日期部分hour()取小时字段。这个方案本身没问题但它要求「时间字段恰好在当天 20 时之后爬取」否则当天的 20 时数据还没生成四个时次凑不齐四个样本该城市会被过滤掉。这个约束在爬虫侧解决当天 20 时 30 分之后再启动爬虫。实践里不少人大白天跑爬虫算出来很多城市日均温为空这是数据时效性问题不是代码 bug。df_avg df_four_points.groupBy(province, city_name, city_code, date) \ .agg( count(col(temperature)).alias(cnt), sum(col(temperature)).alias(temp_sum) ) \ .filter(col(cnt) 4) \ .withColumn(avg_temperature, format_number(col(temp_sum) / 4, 1))cnt 4是校验条件只有 02/08/14/20 四个时次齐了才能算日均温只有 3 条就丢弃。format_number(..., 1)保留一位小数输出形如 18.5。随后对avg_temperature降序排列拿到当日最热城市 Top-N。到这里Spark 端的全部分析逻辑结束。4. 把抽象数据变成可交付结果CSV 合并、JSON 落盘与可视化输入4.1 分区文件的合并与重命名上一章留了一个尾巴——coalesce(1)虽然把分区降为 1但 Spark 写出的文件名仍是随机的part-00000-xxx.csv且会伴随_SUCCESS标记文件。交付课设时应该合并成一个可读的 CSVimport glob, shutil, os part_file glob.glob(output/rain_top/part-*.csv)[0] output_file output/24h_rain_top.csv if os.path.exists(output_file): os.remove(output_file) shutil.move(part_file, output_file) os.remove(output/rain_top/_SUCCESS)shutil.move只是搬运文件真正值得记住的是glob的匹配写法——Spark 输出目录里的part-前缀文件必然存在直接用通配符取第一个即可。这个过程在本地文件系统上没有任何风险换成 HDFS 则应该用hdfs dfs -getmerge来合并不要手动 move。4.2 JSON 落盘与可视化输入气温分析的结果写入 JSON这是为了直接传给 matplotlib 做图A 同学在代码里把 DataFrame 转成 list 后交给可视化脚本df_avg.write.json(output/avg_temperature, modeoverwrite) df_avg_local df_avg.collect() data_for_plot [tuple(row) for row in df_avg_local] print(data_for_plot[:5])collect()会把数据全部拉回 driver5.8 万条的数据量没问题但如果数据量放大到千万级collect 会直接把内存打爆届时应改用limit()或者只拉 Top-N。A 同学的原始项目里气温部分通过collect()转 list 是可行的因为到这一步只有 2412 个城市的 4 行时次记录了。4.3 可视化部分一张降雨 Top 图与气温分布图可视化用 matplotlib 读 CSV 出图不走 Spark 侧因为画图是单机工作Spark Streaming 产生的结果直接给 matplotlib 消费import pandas as pd import matplotlib.pyplot as plt rain pd.read_csv(output/24h_rain_top.csv) top15 rain.head(15) plt.figure(figsize(10, 6)) plt.barh(top15[city_name], top15[rain24h], color#4C72B0) plt.xlabel(24h rainfall(mm)) plt.title(City 24h rainfall TOP 15) plt.gca().invert_yaxis() plt.tight_layout() plt.savefig(result_rain_top15.png, dpi150)这里刻意用pd.read_csv而不是 Spark 再读一遍是因为 Spark 的 write 产物已经在本地文件系统Pandas 读取更直接一整张图 5 秒内出。需要注意invert_yaxis()让降雨量最大的城市显示在顶部否则默认 barh 是反的图形看起来就不专业了。A 同学用图表验证时发现过「最热城市」和「降雨最多城市」完全是两个榜单这说明数据源有效。做课设答辩时把两边的结果合并在一页比单放一堆数字更有说服力。5. 避坑与排查从爬虫到 Spark 结果的 6 个典型问题5.1 Spark 读 CSV 中文乱码或表头变数据现象show()时中文城市名全部变成乱码或第一行 province 出现在数据里。原因CSV 写入时用了utf-8-sig带 BOM编码而spark.read.csv默认按 UTF-8 无 BOM 处理BOM 前缀被当成内容。解决读入时显式指定encodingUTF-8但如果 CSV 是用 Excel 二次编辑后保存的编码可能是 ANSI此时应当先把 CSV 用文本编辑器统一转成 UTF-8。另外确认headerTrue否则第一行会被当数据。提示encodingUTF-8能处理 BOM但如果文件是 GBK 编码必须用encodingGBK不要盲猜。5.2 爬虫跑到一半被限流返回空列表但程序不报错现象日志显示一直在打印城市名但 count 不增长程序像卡死了一样。原因目标站点在一定频率内对连续请求做了限制短时间高频请求后返回空 JSON。程序里if data:判断通过后 count 才自增空数据不会报错循环继续空转。解决每请求一个城市后固定time.sleep(1)慢一点但稳定。另一个办法是加一个连续空响应计数超过 20 次主动退出等 5 分钟再继续。A 同学最初用无 Sleep 版本跑了 500 个城市后陷入假死加上 Sleep 后一口气跑完全部。5.3 降雨量中出现 9999 这种异常值污染排名现象Top 城市降雨量突然出现 3000mm 这种离谱数字。原因传感器故障或数据缺失时气象站会回传特殊值还有一部分是空字符串或--占位符转换成的数值累计后总量异常。解决聚合前强制过滤。col(rain1h) 1000是一种快速规则更稳妥的做法是业务侧确认「单小时降雨量上限」超过上限直接剔除。过滤必须在groupBy之前做过滤时机错了等于没过滤。5.4 日平均气温结果比天气预报高 5 度现象算出来某城市日均温 33℃但当天实际体感没那么热。原因如果爬取时间是下午 14 时数据只包含 02 时和 08 时加上部分 14 时记录20 时记录不存在。此时每天只有 23 个时次被纳入聚合A 同学用cnt 4过滤后直接剔除不完整城市报表上会缺很多城市——缺数据总比错数据好。解决在爬虫侧控制时间当天 20 时 30 分后再启动抓取如果等不到可以在分析时放宽到cnt 3并标注「日均温为三时次近似值」。做课设时直接重跑数据即可。5.5 同一城市名出现在多个省份分组错误现象Top 榜上出现两个「市辖区」或同名县级市数据混在一起。原因全国范围内县市同名很常见只按 city_name 分组会合并不同省份的同名城市导致降雨量虚高。原脚本按province city_name city_code三列分组city_code 是五位站号全国唯一。解决city_code 必须进入groupBy的键组报告里展示城市名时公开「省份-城市」组合避免同名歧义。这和电商订单表必须用order_id sku_id分组是一个道理。5.6 重复爬取导致数据翻倍现象同一个城市出现 48 条记录而不是 24 条总量从 5.8 万变成 11 万。原因爬虫检查了passed_weather_ALL.csv不存在才爬但文件已存在时直接 return如果中间手动删了province.csv但没删天气 CSV代码会跳过全部抓取反过来如果天气 CSV 删了但省份和城市 CSV 还在抓取会从第二步继续不会重复拉省份和城市接口。真正危险的是「部分写入中途退出后重跑」追加模式下文件已有 1000 行重跑又从第 1 个城市开始造成重复。解决加幂等控制。最简单的方式是写完后记录city_index的最大值下次启动前用spark.read.csv读出已有城市编号抓取时跳过。或者每次重跑前直接删除三个 CSV 文件代价是全部重爬。6. 用一张 top 榜单验证全链路正确性课程设计交付到这一步最容易被问的问题是「你怎么知道分析结果是对的」。我自己会走完这套验证手段确定数据管道没问题后再去生成最终图表。第一步抽样还原随机挑一个城市从原始 CSV 里筛选出该城市的 24 条记录手算sum(rain1h)与 Spark 输出的rain24h对比误差应在一分钱以内因为中间做了cast(decimal(10,2))截断允许 0.1 毫米级的舍入差异。验证项手法与预期通过标准总数核对df.count()与wc -l passed_weather_ALL.csv对比一致或相差一行表头Top-N 抽样手动算某市 24h 降雨量与 Spark 输出差 0.1mm气温时次校验df_four_points.groupBy(city_name).count().filter(count4)无或极少输出文件目标目录只存在 1 个 CSV/JSON保证后续读取不出错第二步异常值回查把rain1h 1000但被过滤掉的记录单独拉出来看是不是传感器或爬虫解析异常确认过滤规则没误杀。这个习惯保留到现在每次跑完聚合我都会先看一下被过滤记录的总数和形态。第三步数据集级的 sanity check降雨量 Top 榜应在多雨省份居多日均温 Top 榜应符合当天的气候常识。如果榜单里出现沙漠城市降水量全国第一说明要么爬虫数据串了城市编号要么过滤规则写错。A 同学当时就是靠这一步发现某城市编码错位导致数据串行由此养成了「榜单必须和现实对得上」的验证习惯。最终交付物包括24h_rain_top.csv、avg_temperatureJSON、两张 matplotlib 图、以及完整的爬虫和 spark 分析脚本。用离线数据复现全程 20 分钟以内答辩现场可以直接从 CSV 重新触发分析不需要再跑一次爬虫。从那以后我每次做 Spark 课设都会强制走一遍「抽样还原 异常值回查 常识比对」这三步避免在讲台上对着一个排名靠前的错误数据集硬讲结论。希望帮到你这套流程直接搬去用就行。本文还有配套的精品资源点击获取