简介：面向大数据期末课设的Spark气象数据处理与分析完整报告，适合需要完成Spark实战项目、掌握气象数据分析全流程的大数据专业学生参考。项目基于中央气象台2412个城市、57888条真实天气数据，完整呈现从JSON异步数据爬取、CSV存储、Spark DataFrame清洗聚合到matplotlib可视化的实现过程，并给出省份、城市、天气三级接口的解析思路与爬虫源代码。报告按课程设计标准撰写，章节涵盖项目背景、实验环境、数据来源、数据获取、数据分析与可视化，其中重点演示了分组聚合、缺失异常值处理、窗口函数等Spark常用操作，可作为课程报告或答辩展示的实用模板。内含1个docx文档（744KB），已有13405人学习，适合大数据入门者对照演练。

1. 基于Spark的气象数据处理与分析：一份能直接复用的大数据期末课设

每年期末，总有同学卡在同一个问题上：Spark课设到底做什么才能既体现技术含量，又能在两周内跑完？这份基于Spark的气象数据处理与分析课设项目，走的就是一条非常务实的路线——爬取全国2412个城市、57888条整点天气数据，用Spark SQL完成24小时累计雨量和日均气温的聚合计算，再用matplotlib把结果画出来。整套流程覆盖了数据获取、清洗、分析、可视化四个环节，技术栈就是Spark DataFrame API加Python，没有绕圈子，也没有堆砌高深算法。适合大数据专业期末课设、Spark入门实战，以及想了解“用Spark做气象数据处理”完整链路的人。下文我按实际落地顺序，把环境、爬虫、预处理、分析和踩坑点逐一拆开讲。

2. 环境与数据获取：从三级URL到57888条天气数据的爬取链路

2.1 实验环境选型与组件安装

这个课设的实验环境是一个典型的单机Spark开发环境，不需要搭建集群。我复现时使用的组件版本如下表所示：

有一点值得注意：Spark 2.4.0与Python 3.9的组合在实际使用中会有兼容性警告，网上很多资料说Spark 2.4.0不支持Python 3.8以上的版本，实际跑起来Spark SQL基础功能问题不大，但如果你用到pandas UDF或新的API，建议直接换Spark 3.x。课设场景下沿用2.4.0没有问题，只要别在它上面做太新奇的特性就行。

装完Spark和Jupyter之后，还需要给Python补上可视化的依赖：

sudo apt-get install python3-matplotlib sudo apt-get install python3-tk

python3-matplotlib是绘图库本体，python3-tk是matplotlib在本地窗口显示图表时需要的Tkinter后端。如果你用的是无显示器的服务器，还应在代码开头加上matplotlib.use('Agg')，否则绘图时找不到显示设备会直接报错。Ubuntu 16.04的apt源里自带这两个包，优先用它装，比pip装更省事。

2.2 三级URL接口设计：先拿省份，再拿城市，最后拿天气

气象数据的来源是某公开气象服务站点。通过浏览器开发者工具观察网络请求，能发现数据不是渲染在HTML页面里的，而是通过异步请求直接返回JSON字符串。核心的请求路径可以归纳为下面三条：

三级接口是典型的嵌套依赖关系：省份编码必须从第一级接口的返回里取，城市编号必须从第二级接口的返回里取，第三级接口才能拿到真正的气象数据。比如某个省在省份接口里的编码是“ABC”，请求城市列表时要带上这个编码；返回的城市数据里又有类似“58847”的城市编号，把它拼到第三级URL末尾才能拿到该城市24小时整点数据。爬虫就是要顺着这条链子把全国城市依次走一遍。

2.3 爬虫代码实现与CSV分段落盘

爬虫部分原课设报告用的是Python 2时期的urllib3.Request写法，但urllib3库里并没有Request类，直接跑会报AttributeError。我改成Python 3标准库的urllib.request，逻辑保持一致：

import urllib.request import json import csv import codecs import os import time class WeatherCrawler: def __init__(self, base_url): self.base_url = base_url def fetch_json(self, path): url = self.base_url + path req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode('utf-8')) def write_csv(self, file_name, rows, write_header=False): if not rows: return with open(file_name, 'ab+') as f: f.write(codecs.BOM_UTF8) writer = csv.DictWriter(f, fieldnames=rows[0].keys()) if write_header: writer.writeheader() writer.writerows(rows)

fetch_json里设置了User-Agent请求头，避免站点拒绝默认的Python urllib请求。timeout=10是超时控制，防止某个城市接口卡住导致整个爬虫挂起。write_csv里写入codecs.BOM_UTF8是为了让Excel直接打开CSV时中文不乱码；write_header参数控制是否写表头，因为分段落盘时只有第一个文件块需要写表头。

下面是三级数据获取的完整调度逻辑：

def get_provinces(self): if os.path.exists('input/province.csv'): return self.read_csv('input/province.csv') provinces = self.fetch_json('/f/rest/province') self.write_csv('input/province.csv', provinces, write_header=True) return provinces def get_cities(self): if os.path.exists('input/city.csv'): return self.read_csv('input/city.csv') cities = [] for province in self.get_provinces(): code = province['url'].split('/')[-1].split('.')[0] cities.extend(self.fetch_json('/f/rest/province/' + code)) time.sleep(1) self.write_csv('input/city.csv', cities, write_header=True) return cities def get_passed_weather(self): cities = self.get_cities() passed_weather = [] count = 0 for city in cities: data = self.fetch_json('/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 and count > 0: self.write_csv('input/passed_weather_ALL.csv', passed_weather, write_header=(count == 50)) passed_weather = [] if passed_weather: self.write_csv('input/passed_weather_ALL.csv', passed_weather, write_header=(count <= 50))

这段代码有两个关键设计。第一个是幂等：province.csv和city.csv已存在时直接读取，全量天气数据已经落盘后重新运行也不会重复爬取；想刷新数据就手动删掉这三个文件。第二个是每爬50个城市就落盘一次——全国2412个城市，每条数据有多个整点记录，如果全部堆在内存里等到最后再一次性写文件，爬半小时内存早就爆了。分段写入之后，即使中间某个城市请求超时导致中断，已爬的部分也不会丢。

3. Spark SQL数据预处理：Schema推断、类型转换与异常过滤

3.1 SparkSession初始化与CSV读取

数据爬下来之后，下一步就是用Spark读进来。这里用的是Spark SQL中的DataFrame API，而不是RDD——DataFrame有类型信息，SQL风格表达清晰，而且Spark默认的Catalyst优化器会自动优化执行计划，对课设这种量级的数据完全够用。

from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("WeatherAnalysis") \ .master("local[4]") \ .getOrCreate() df = spark.read.csv( "input/passed_weather_ALL.csv", header=True, inferSchema=True )

master("local[4]")表示本地模式用4个线程并行执行，对六万条数据来说绰绰有余。header=True把CSV第一行作为列名，inferSchema=True让Spark扫描数据后自动推断每一列的类型。这样temperature和rain1h会被识别成double，后续做数值计算不用手动转换。

但inferSchema有一个坑：如果CSV某列的前面一部分行是缺失值或非数字字符串，Spark可能把整列推断成string，后面用比较运算符直接报错。稳妥的做法是在读取后打印Schema检查一遍：

df.printSchema() df.show(5, truncate=False)

printSchema()输出的每一行格式是“列名 类型”，一眼就能看出有没有推断错误。show(5, truncate=False)则是查看前5条原始数据，truncate=False确保长字段不会被截断，方便直接观察time字段的格式。

3.2 字段选择与数值类型转换

原课设报告里有一个很容易被忽略的细节：CSV经过Python爬虫写出后，即使Spark用inferSchema读入，某些列的类型仍是字符串。更稳妥的做法是在分析前显式做一次类型转换，把注意力集中在真正需要的列上：

from pyspark.sql.functions import col df_rain = df.select( "province", "city_name", "city_code", "rain1h" ).filter( col("rain1h").cast("double") < 1000 )

select只取四个字段，把数据宽度压缩到最小；col("rain1h").cast("double")强制把降雨量转成数值型，这样即使某一行是空字符串也不会让整个作业报错，而是转为null。过滤条件< 1000是排除异常值——正常城市24小时累计降雨量不可能超过1000毫米，一旦出现这种记录，基本是站点数据采集故障或爬取时拿到的脏数据。

3.3 时间字段解析与温度数据准备

温度分析需要把time字段拆成“日期”和“小时”两部分，这就要用到pyspark.sql.functions里的两个函数：

from pyspark.sql.functions import date_format, hour df_temp = df.select( "province", "city_name", "city_code", "temperature", "time" ).withColumn( "date", date_format(col("time"), "yyyy-MM-dd") ).withColumn( "hour", hour(col("time")) )

date_format(col("time"), "yyyy-MM-dd")从时间戳里截取日期部分，hour(col("time"))提取小时数值。这里的前提是CSV里time字段能被Spark识别为时间类型。如果Spark读进来之后time是字符串，这两个函数会返回null。遇到这种情况，先看原始数据的time字段格式，再决定是加一个to_timestamp(col("time"), "yyyy-MM-dd HH:mm:ss")转换，还是直接用字符串截取函数处理。

整个预处理阶段的目标，就是得到两张干净的中间表：一张是带rain1h数值的df_rain，一张是带date和hour字段的df_temp。两张都是懒加载的DataFrame，真正触发计算的是后续的groupBy聚合操作。

4. 两个核心分析任务：24小时累计雨量与四时次日均气温

4.1 累计降雨量分析：groupBy、agg与全局排序

降雨量分析的思路很直接：按省份和城市分组，对rain1h字段求和，得到每个城市过去24小时的累计雨量，然后降序排列取Top-N。Spark代码是这样的：

df_rain_sum = df_rain.groupBy( "province", "city_name", "city_code" ).agg( {"rain1h": "sum"} ).withColumnRenamed( "sum(rain1h)", "rain24h" ).sort( col("rain24h").desc() )

agg({"rain1h": "sum"})是传入字典的写法，键是列名，值是聚合函数，结果列名自动生成为sum(rain1h)，所以紧接着用withColumnRenamed重命名成rain24h，后面引用起来更直观。.sort(col("rain24h").desc())是分布式环境下的全局排序，Spark会做一次shuffle把所有分区数据按雨量倒排，对2400多个城市的数据量来说代价很小。

排序之后，head(10)可以直接取到雨量最大的十个城市，这个动作会触发Spark真正执行前面的所有转换步骤：

top_rain = df_rain_sum.head(10) for row in top_rain: print(row["city_name"], row["rain24h"])

注意head(10)返回的是一个Row对象列表，每个Row就像是Python里的字典，用row["列名"]取值。这一步执行时，Spark会把懒加载的整个血缘关系链从CSV读取一路算到排序，所以如果前面某一步写错了，报错是在这里炸出来，而不是在groupBy那里。

4.2 日均气温分析：四时次筛选与聚合计算

日均气温这一段是项目里最有气象业务味道的地方。根据《地面气象服务观测规范》，日平均气温取的是02时、08时、14时、20时四个时次气温的平均值，而不是24小时整点气温的算术平均。这是气象行业的规范，直接决定了代码里的过滤条件：

from pyspark.sql.functions import count, sum as spark_sum, format_number df_4point = df_temp.filter( col("hour").isin([2, 8, 14, 20]) ) df_avg = df_4point.groupBy( "province", "city_name", "city_code", "date" ).agg( count("temperature").alias("cnt"), spark_sum("temperature").alias("temp_sum") ).filter( col("cnt") == 4 ).withColumn( "avg_temperature", format_number(col("temp_sum") / 4.0, 1) ).select( "province", "city_name", "city_code", "date", "avg_temperature" )

整个流水线的关键是先计数再求均值：count("temperature")看看每个城市在目标日期里是否四个时次的数据都齐了，只有cnt == 4的记录才参与平均计算。如果某个城市某天只有3个时次的数据，直接算平均气温会和真实值有偏差，过滤掉才是严谨的做法。format_number(..., 1)把结果格式化成带一位小数的字符串，方便直接用于可视化展示。

这个查重逻辑在课设报告里已经说得很细，但我复现时发现一个数据层面的问题：爬虫是在白天某个时刻运行的，当天20时还没到，那么当天20时的记录根本不存在。原报告也提示了“建议在当天20时30分之后再爬取数据”，这说明分析结果的质量高度依赖爬取时机。

4.3 cache与coalesce：性能优化和输出控制

两个分析任务在最后都调了cache()和coalesce(1)，这一步值得展开讲：

df_rain_sum.cache().coalesce(1).write.csv("output/rain_sum", header=True) df_avg.coalesce(1).write.json("output/avg_temp", header=False)

cache()把前面经过groupBy和sort的中间结果物化到内存里，这样后面如果既要做head(10)又要write.csv，Spark不用从头再算两遍。coalesce(1)是把分区数量压缩到1个，这样CSV输出只有一个文件，而不是默认的几十个part-00000碎片，后续直接读这个单文件就行。

但coalesce(1)不是免费的。它本质上是一次结果重分区，把所有分区的数据拉到同一个分区再写出，数据量大时会产生单点瓶颈。雨量Top-N分析加coalesce(1)没问题，但如果以后换成每天几十GB的气象数据，正确做法是保留分区数并写成parquet格式，不要强行合并成单文件。这个边界要拎清楚。

5. 常见问题与排查：爬虫、Spark和可视化里的五个坑

5.1 Python 3下运行报错：sys.setdefaultencoding不存在

现象：把课设原爬虫代码直接丢到Python 3环境运行，启动就报AttributeError: module 'sys' has no attribute 'setdefaultencoding'。

原因：sys.setdefaultencoding('utf8')是Python 2时代处理默认编码的命令，Python 3的默认编码就是UTF-8，这个函数已经被移除。原课设写的是reload(sys)和sys.setdefaultencoding('utf8')，这是典型的Python 2代码。

解决：直接删掉这两行。如果担心中文编码有问题，用Python 3在打开文件时指定encoding='utf-8'就够了，不需要再动全局默认编码。这个坑几乎是每届课设必现，看到这个报错就知道代码是从旧版本迁移过来的。

5.2 爬虫结果CSV用Excel打开中文乱码

现象：passed_weather_ALL.csv在终端和代码里读着都正常，用Excel双击打开时中文全部变成乱码，省份名和城市名完全不可读。

原因：CSV文件没有BOM头。Python默认用UTF-8写文本文件不掺BOM，而Excel的CSV解析器在识别非UTF-8编码时，倾向用ANSI解码中文，结果就是乱成一团。原报告里在write_csv中写入codecs.BOM_UTF8，就是为了解决这个问题。

解决：写入CSV前先写一次codecs.BOM_UTF8，这是给Excel看的编码标记。注意每次打开文件追加写入时都要判断文件是否已存在，如果已存在就不要重复加BOM，否则表头里会多出隐藏的非法字符。

5.3 Spark读CSV后数值列变成字符串，过滤直接报错

现象：df.filter(col("rain1h") < 1000)执行时抛出分析异常，提示无法用字符串和数字比较；打印Schema发现rain1h列的类型是string而非double。

原因：inferSchema=True不是万能的。如果CSV的前若干行里rain1h字段是空值或占位符（比如--），Spark在推断类型时会认为整列是字符串。尤其是爬虫写CSV时把数字转成了字符串，空字段就是空字符串，更容易触发误判。

解决：最直接的办法是用col("rain1h").cast("double")做显式转换，再配合filter过滤。另一种做法是给read.csv显式传schema参数，例如StructType里指定rain1h为DoubleType()，彻底跳过推断机制。从工程习惯来说，显式指定Schema比inferSchema更可靠，但课设阶段写清楚cast就够了。

5.4 日均气温结果城市数大幅缩水，只有几百个城市

现象：计算完日均气温后，输出结果里的城市数远少于数据源里的2412个城市，少了一半还多。检查原始数据，发现每个城市确实有整点气温数据。

原因：filter(col("cnt") == 4)把只拥有部分时次记录的城市全部剔除了。爬虫运行时间是白天，当天20时的数据不存在；有些城市02时或08时的记录缺失，凑不齐四时次的都被过滤掉了。

解决：四种选择。一是按原报告建议，在当天20时30分之后再爬取，保证当天四个时次都齐；二是把cnt == 4放宽到cnt >= 3并在报告里说明“基于三时次近似计算日均气温”，适合只想跑通流程的情况；三是查看缺失时次的分布规律，找出是哪几个时次最容易缺；四是修改爬虫逻辑，补充历史数据接口而不是只爬最近24小时。课设报告采用的是第一种，输出的质量是最高的。

5.5 结果目录里出现一堆part文件，没有单个CSV文件

现象：执行df_rain_sum.write.csv("output/rain_sum")后，打开目录看到的是part-00000-xxx.csv、part-00001-xxx.csv十几个文件，而不是想象中的rain_sum.csv。

原因：Spark分布式写入时，每个分区各自写一个文件，分区数就是输出文件数。默认情况下shuffle之后分区数较多，自然会产生多个part文件。这不算错误，但对后续人工查看数据很不方便。

解决：写之前加coalesce(1)，强制把分区数减到1，输出就只有一个part文件了。更省事的办法是不加coalesce(1)，读取时直接spark.read.csv("output/rain_sum")读整个目录，Spark会自动合并多个文件的内容。两种方式我都用过，课设评分时如果要求提交“干净的CSV文件”，就选coalesce(1)。

6. 结果验证与可视化输出：用matplotlib把分析结果变成图表

数据分析算完不算结束，还要验证结果正确，然后再画图。验证有一个很土但很有效的办法——抽样对账。我从df_rain_sum.head(10)里随机挑了三个城市，找到它们在passed_weather_ALL.csv里的所有rain1h记录，手工在Excel里求和，和Spark算出的rain24h对比。三次都对得上，说明聚合逻辑没问题。这一步强烈建议保留，它能把“代码跑通”和“结果正确”区分开，答辩时也更有底气。

可视化部分代码比较简短，用matplotlib画横向条形图展示雨量Top-10城市：

import matplotlib.pyplot as plt top = df_rain_sum.head(10) names = [row["city_name"] for row in top] values = [float(row["rain24h"]) for row in top] plt.rcParams["font.sans-serif"] = ["SimHei", "WenQuanYi Zen Hei"] plt.rcParams["axes.unicode_minus"] = False plt.figure(figsize=(10, 6)) plt.barh(names[::-1], values[::-1]) plt.xlabel("24小时累计降雨量（mm）") plt.title("过去24小时累计降雨量 Top10 城市") plt.tight_layout() plt.savefig("output/rain_top10.png", dpi=150)

plt.rcParams两行是解决中文字符显示成方块的设置，SimHei是Windows下的黑体，Linux下没有时就回退到WenQuanYi Zen Hei。names[::-1]和values[::-1]是把列表倒序，让雨量最大的城市显示在条形图最上方，符合阅读习惯。savefig的dpi=150保证图片在答辩投屏时不模糊。

气温分析的结果可以用类似思路画一张散点图或直方图，查看全国城市日均气温的分布形态。到这一步，从爬虫、清洗、聚合到出图，整条链路就闭环了。

想起那次复现后半段，我发现某个城市计算出日均气温为97度，一查是20时的气温字段爬到了异常值，温度列混入了噪点数据。从那以后，我每次做数据类课设都会强制走一遍“抽样对账”再进入可视化，这个习惯帮我挡掉了至少三次答辩翻车。希望这份基于Spark的气象数据处理与分析课设梳理，能帮你也少踩几个同样的坑。

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