简介：一份基于Spark框架的气象数据处理与分析课程设计报告，适合高校大数据相关专业学生、课程设计参与者及对气象数据分析感兴趣的读者，用于学习和掌握Spark处理结构化数据的方法。报告完整覆盖从数据获取到可视化展示的流程：以中央气象台网站为数据源，使用Python爬虫采集近24小时各城市整点天气数据，覆盖2412个城市、57888条记录，并对缺失或异常时间点做过滤处理；随后通过Spark SQL计算城市累计降雨量与日平均气温，利用matplotlib绘制可视化图表。资源还介绍了Linux Ubuntu 16.04、Python 3.9、Spark 2.4.0、Jupyter Notebook等实验环境的搭建，以及Spark在大数据领域的优势与适用场景。资源为单个docx文档，约724KB，内含项目背景、实验环境、数据获取、数据分析、数据可视化及总结等章节，并附有可参考的爬虫源代码。已有169人学习下载，可作为课程设计报告模板和Spark实战入门参考，对完成气象类大数据课设具有实用价值。

1. 从气象课设读 Spark 的取舍：57888 条天气数据够讲明白什么事

一份 2412 个城市、57888 条记录的课程设计，单看规模不大，真正有价值的不是数据量，而是它把一个完整链路走通了：爬虫按「省份 → 城市 → 整点天气」三级接口取数，存成 CSV 后再交给 Spark SQL 做分组聚合，算出 24 小时累计降雨量和按四时次标准计算的日平均气温，最后落到 matplotlib 可视化。对正在做课设、或者刚接触 Spark 生态想找个真实场景练手的人来说，这份资源最大的价值在于「能看到 2.4.0 这代 Spark 在真实环境里的脾气」——哪些参数能调、哪些异常必须过滤、数据时次不齐对结果影响多大。我拆完这个项目后最大的感受是：Spark 的坑很少出在计算本身，基本都在数据进 DataFrame 之前和结果落盘之后。

2. 实验环境与数据链路：Ubuntu 16.04 上的标准物料清单

2.1 版本组合的兼容性边界：Spark 2.4.0 与 Python 3.9 的取舍

原文报告的实验环境是 Ubuntu 16.04、Python 3.9、Spark 2.4.0、Jupyter Notebook。这套版本组合单独看都没问题，但如果你打算在自己机器上复现，有两个点需要提前想清楚。

第一，Spark 2.4.0 是 2018 年底发布的版本，官方对 Python 3.9 的正式支持其实是从 Spark 3.0 开始的。2.4.0 用 Python 3.9 跑 pyspark 大概率能启动，但某些内建函数在极端情况下会踩到字节码兼容问题。我一般建议：如果完全照抄报告环境，可以退到 Python 3.6 或 3.7；如果坚持用 Python 3.9，就把 Spark 升到 3.1 以上。当然，课设场景下 Spark 2.4.0 + Python 3.9 是可以跑通报告里这两段分析代码的，不用过度担心。

第二，Ubuntu 16.04 自带的 Python 3.5 跟报告要求的 3.9 不一致，需要单独装。检查环境的命令如下：

java -version python --version spark-shell --version

三个命令依次确认 JDK（Spark 2.4 要求 Java 8）、Python 解释器、Spark 版本。其中 java 版本最容易出问题，因为无论你从官网下载的 Spark 二进制包还是用 apt 装的 openjdk，只要 java 版本不对，spark-shell 启动时会直接报Unsupported class file major version之类的错误。

这代 Spark 对环境的敏感程度，比你想的高。它不像现在很多框架那样自带完整的运行时，而是要自己组装 Java、Scala、Python 三套环境的交集。报告里写清楚了版本，你复现时最好一字不差地照着来，等跑通后再考虑升级。

2.2 数据来源的三级接口结构：JSON 异步返回与 URL 规律

数据从中央气象台官网提供的 REST 接口拿，不是爬 HTML 页面。打开任意一个城市的天气页面，在 Chrome 控制台里能看到网络请求列表，接口直接返回 JSON。这和传统的「下载 HTML → 用 BeautifulSoup 解析 DOM」完全不同，省掉了页面解析环节。

三个接口的对应关系如下表所示：

以福州为例，先从省份接口拿到福建的编码ABJ，再拼接出福建的城市列表接口拿到福州的城市编号58847，最后用passed/58847拿到 24 条整点数据。这是一个典型的三级联动爬取，逻辑简单，但层级关系必须梳理清楚。

值得注意的一点是：城市接口返回的城市列表里，code字段不只是数字编号，有些城市编号是独立的站点代码，直接拼到passed/后面就行。但如果某个城市没有自动气象站或者站点数据未接入，接口会返回空数组，这一步要在代码里做过滤，不然后续处理会频繁报错。

2.3 CSV 字段设计与落盘策略：为什么不用 JSON 直接喂给 Spark

报告里爬虫拿到 JSON 后没有直接存 JSON，而是转成了 CSV，并且拆成三个文件：province.csv、city.csv、passed_weather_ALL.csv。这个选择很实用。

JSON 虽然能被spark.read.json()直接读取，但嵌套结构不利于肉眼排查异常；CSV 是扁平化的，每一行就是一条整点记录，字段含义直白，用 Excel 打开也能直接检查。报告在写入 CSV 时做了三个处理：加 UTF-8 BOM（兼容 Excel 打开中文不乱码）、每爬满 50 个城市追加写一次（防止内存堆积）、在每条记录里冗余写入省份名和城市名（避免后续 join 省份表）。核心字段只有 7 个：

我在复现时注意到一个细节：rain1h和temperature在 JSON 原始返回里是字符串类型，所以后面 Spark 分析时第一步必须先cast成数值型。这不算 Bug，但如果你直接对字符串列做sum，Spark 会把它当 concat 处理或者直接报错，是新手最容易踩的坑。

3. 爬虫实现：JSON 接口直取与断点续爬的关键细节

3.1 三级请求链路的代码骨架：省份、城市、天气三个函数

爬虫逻辑不复杂，核心就是用 urllib 内置模块请求 JSON 接口，然后按层级往下取数。原报告的写法里有几个函数名混用了 urllib3，实际上标准库模块是urllib.request，接口长这样：

import urllib.request import json import csv import codecs import os import time class Crawler: def get_html(self, url): # 标准库写法，不需要额外安装 requests req = urllib.request.Request(url) resp = urllib.request.urlopen(req, timeout=10) return resp.read() def parse_json(self, url): # 接口直接返回 JSON，不需要解析 HTML obj = self.get_html(url) if obj: return json.loads(obj) return [] def get_provinces(self): # 第一级：省份列表 prov_file = 'input/province.csv' if not os.path.exists(prov_file): provinces = self.parse_json('http://www.nmc.cn/f/rest/province') self.write_csv(prov_file, provinces) else: provinces = self.read_csv(prov_file) return provinces

逻辑说明：get_provinces()做了一个「文件存在就直接读、不存在才重新爬」的判断，这是整个断点续爬设计的基础。第一次运行会请求省份接口并落盘，之后每次运行直接读本地 CSV，既省时间又避免反复请求对方服务器。parse_json()返回的是 Python list，每个元素是一个 dict，字段里既有省份编码又有中文名，后续通过它拿到ABJ这样的编码去拼接城市接口。

参数说明：timeout=10是连接超时时间，单位为秒。气象接口虽然稳定，但可能存在某个城市节点响应慢的情况，不设超时会导致爬虫卡死。如果你在学校网络环境跑，timeout可以放宽到 15，如果只是课堂演示，10 秒够用。

省份数据拿完后，下一层是城市列表：

def get_cities(self): city_file = 'input/city.csv' if not os.path.exists(city_file): cities = [] for province in self.get_provinces(): # 省份数据里的 url 字段类似 "/f/rest/province/ABJ.html" # 取最后一个 / 到 . 之前的片段作为编码 prov_code = province['url'].split('/')[-1].split('.')[0] part = self.parse_json( 'http://www.nmc.cn/f/rest/province/' + prov_code ) cities.extend(part) self.write_csv(city_file, cities) else: cities = self.read_csv(city_file) return cities

逻辑说明：province['url']是省份接口返回的完整链接，形如/f/rest/province/ABJ.html。这里两次 split 的目的就是把三位编码ABJ抠出来，然后拼到固定前缀后面。注意part是当前省份下所有城市的列表，要用extend合并，不能用append，否则列表会变成嵌套结构。

3.2 批量写盘与进度控制：每 50 个城市一次落盘

爬完城市列表后就到了最耗时的环节——逐个城市请求passed/接口。报告明确写「每爬取 50 个城市的数据后，就会进行一次保存」，这个策略很聪明，因为全网 2400 多个城市逐条请求，加上每次 1 秒的延时，总时长在半小时以上，如果全部攒在内存里最后一次性写盘，中途任何一次网络异常都意味着之前的数据白爬。

def get_passed_weather(self): weather_file = 'input/passed_weather_ALL.csv' if os.path.exists(weather_file): print('weather file already exists, skip') return passed_weather = [] count = 0 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['province'] = city['province'] item['city_name'] = city['city'] item['city_code'] = city['code'] item['city_index'] = str(count) passed_weather.extend(data) time.sleep(1) # 控制请求频率，避免被封 # 每 50 个城市写一次盘，防止内存堆积 if count % 50 == 0: if count == 50: self.write_header(weather_file, passed_weather) else: self.write_row(weather_file, passed_weather) passed_weather = [] # 收尾：不足 50 的余数也要落盘 if passed_weather: self.write_row(weather_file, passed_weather)

逻辑说明：city['code']直接拼在passed/后就是城市 24 小时天气接口。拿到返回后，代码把城市名、省份名、城市编号都冗余塞进每一条 weather 记录里，这样后续分析就只需要读一个文件，不用再 join 另外两张表。time.sleep(1)是必要的，气象网站虽然不一定会封 IP，但 2400 多个城市连续高频请求，礼貌性限速能避免不必要的麻烦。

write_header和write_row的区别在于：第一次写 CSV 需要写表头，之后只追加数据行。count == 50时恰好凑满第一批，所以用带表头的写法。这里的count % 50 == 0是边界条件的关键——如果写成了count > 0，每个城市都会触发一次写盘，性能会非常差。

3.3 断点续爬与数据校验：爬一半中断了怎么办

这个爬虫有一个很实用的特性：天然支持断点续爬。因为get_provinces()和get_cities()都做了「文件存在就不重新爬」的判断，只有get_passed_weather()是无条件全量爬的。如果你在中途断了网，只需要把passed_weather_ALL.csv删掉重新运行，省份表和城市表会直接读本地文件，不会重复请求前两级接口。

但注意，get_passed_weather()本身没有做已爬城市的去重。也就是说，重新爬会全量跑一遍 2400 多个城市，而不是从断点接着爬。这在课设里可以接受，反正半小时能跑完；但如果你要大规模采集，建议在代码里先读一遍已存在的 CSV，把city_code集合拿出来，在循环开头加一个if city['code'] in done_set: continue的判断。

另外，拿到 CSV 后最好做一次完整性检查。简单做法是读取行数，看是否接近 57888 条（2412 城市 × 24 整点）。如果差得很多，大概率是有部分城市返回的空列表太多了，需要重新爬或者接受这个不完整的数据集继续分析。报告里的处理方式是「部分城市部分时间点数据存在缺失或异常」，明确承认了数据不完美，这种态度在课设里反而是加分项。

4. Spark SQL 核心分析：24 小时累计雨量与日平均气温

4.1 读取 CSV 与类型转换：字符串字段必须先 cast

爬虫落盘后，进入 Spark 分析阶段。报告用的是 Spark SQL 的 DataFrame API，不是 RDD 算子。整个过程分两个任务：先算各城市 24 小时累计降雨量，再算各城市当日平均气温。

第一步是创建 SparkSession 并读取 CSV：

from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum, avg, date_format, hour, format_number spark = SparkSession.builder \ .appName("WeatherAnalysis") \ .master("local[*]") \ .getOrCreate() df = spark.read.csv( "input/passed_weather_ALL.csv", header=True, inferSchema=True ) df.printSchema()

逻辑说明：header=True告诉 Spark 第一行是表头；inferSchema=True让 Spark 自动推断每列类型。但要注意，即使开了inferSchema，rain1h和temperature也有大概率被推断成字符串——因为原始 JSON 里这两个字段的值是带引号的，加上部分行还混着空字符串，Spark 的类型推断在这种脏数据面前会保守地选择 string。

参数说明：master("local[*]")表示用本机所有 CPU 核跑，课设数据只有 5 万多行，单机完全够用。如果在集群上提交，这里要改成 yarn 模式并去掉master配置，由spark-submit参数指定。

类型转换正确姿势：

df_clean = df.select( col("province"), col("city_name"), col("city_code"), col("rain1h").cast("double"), # 降雨量转数值 col("temperature").cast("double"), # 气温转数值 col("time") ).filter( col("rain1h").isNotNull() & col("rain1h") < 1000 )

逻辑说明：cast("double")是 Spark 的 Column 方法，把字符串列转成双精度浮点型。filter(col("rain1h") < 1000)是报告里明确写的异常值过滤——某些站点异常情况下 rain1h 会返回 9999 这样的巨值，如果不过滤，一个城市 24 小时累计降雨量会被拉到几千毫米，完全没有意义。

4.2 24 小时累计降雨量：groupBy + agg 的分组聚合写法

数据清洗完毕后，累计降雨量的计算就是标准的分组聚合：

df_rain_sum = df_clean.groupBy( "province", "city_name", "city_code" ).agg( sum("rain1h").alias("rain24h") ).sort( col("rain24h").desc() )

逻辑说明：groupBy按省份、城市名、城市编号三个字段分组，这里把city_code也加进去是因为不同省份可能存在同名城市。agg(sum("rain1h"))对每个城市的降雨量求和，新列命名为rain24h。最后sort按降雨量降序排列，得到的是全国 24 小时累计降雨量 Top N 列表。

到这里计算已经完成，但 Spark 是惰性求值的——上面这段代码只是构建了执行计划，并没有真正跑。只有当你调用head()、collect()或write.csv()这些 action 操作时，作业才会真正提交执行。

继续完成落盘和取 TopN：

df_rain_sum.cache() # 缓存中间结果，避免重复计算 # 取前 20 个城市用于可视化 top20_rain = df_rain_sum.head(20) # 合并为 1 个分区再写入，确保只生成一个 CSV 文件 df_rain_sum.coalesce(1).write.csv( "output/rain_sum_result", header=True, mode="overwrite" )

逻辑说明：cache()把计算结果缓存在内存里。这里能缓存成功的关键在于前面已经做了一次 action（head(20)），Spark 把结果物化到内存，后续write.csv才能直接复用，不会再从源头重新算一遍。coalesce(1)把分区合并成 1 个，这样写出的 CSV 只有一个文件而不是散落几十个小块。课设报告里给一个文件是合理的，方便老师直接打开看；生产环境千万不能用，因为单分区会丢掉并行度。

4.3 日平均气温：四时次筛选与分组求均值

日平均气温的计算比降雨量麻烦，因为不是简单的 24 条数据取平均。报告里严格遵循《地面气象服务观测规范》，日平均气温取 02 时、08 时、14 时、20 时四个时次的平均值。

from pyspark.sql.functions import date_format, hour df_temp = df_clean.select( col("province"), col("city_name"), col("city_code"), col("temperature"), date_format(col("time"), "yyyy-MM-dd").alias("date"), hour(col("time")).alias("hour") ).filter( col("hour").isin([2, 8, 14, 20]) )

逻辑说明：date_format(col("time"), "yyyy-MM-dd")从完整时间字符串里截出日期部分，hour()提取小时字段。原始数据里每条记录的时间是整点，分秒都是 0，这两个函数能干净地拆出需要的维度。isin([2, 8, 14, 20])是等值过滤，把非四时次的记录全丢掉。

但是光过滤还不够，报告里加了一步很关键的校验：

df_avg_temp = df_temp.groupBy( "province", "city_name", "city_code", "date" ).agg( count("temperature").alias("cnt"), sum("temperature").alias("temp_sum") ).filter( col("cnt") == 4 ).select( col("province"), col("city_name"), col("city_code"), col("date"), format_number(col("temp_sum") / col("cnt"), 1).alias("avg_temperature") ).sort( col("avg_temperature").desc() )

逻辑说明：这一句是整个气温计算的核心。先按城市和日期分组，count("temperature")统计每个城市当天实际有几个时次的数据；然后filter(col("cnt") == 4)强制要求四个时次全部齐了才参与计算。为什么要这一步？因为爬虫开始时间和城市数量太多，有的城市数据可能只覆盖到 3 个时次，直接求平均会让结果偏低，体现不出真实日均温。format_number(..., 1)是格式化函数，保留一位小数。

报告里特别建议在当天 20 时 30 分之后再爬数据。原因是 20 时的整点数据一般在 20 时过后的某个时间点才稳定上传，如果提前爬，最后一个时次大概率缺失，会直接导致该城市被cnt == 4过滤掉。这是时间敏感型场景，和 Spark 本身无关，但实际复现时最容易栽在这。

气温结果写成 JSON：

df_avg_temp.coalesce(1).write.json( "output/avg_temp_result", mode="overwrite" )

报告里降雨量结果用write.csv落盘，气温结果用write.json。这个选择本身没有功能差异，主要看后续用在哪：CSV 方便 Excel 打开和二次导入，JSON 更结构化、方便 Web 前端直接消费。我在实际做的时候会把两个都出 CSV，因为课设报告插入表格截图比 JSON 截图直观。

4.4 性能细节：cache 与 coalesce 的使用时机

这两个算子单独拆开讲一下，因为它们是这份代码里最能体现 Spark 思维方式的部分。

cache()什么时候用？当你有一个庞大的血缘关系链（DataFrame 经过多次 transform），并且后续有多个 action 要基于同一个中间结果执行时，缓存能让第二个 action 不再重新推导。但这里要注意：第一次 action 触发时 Spark 才会真正执行缓存，所以在cache()之后先调一次head()或count()预热，再执行write操作，才能体会到提速。

coalesce(1)什么时候用？只在最终结果需要「单文件」时用。Spark 写入的分区数与输出文件数完全对应，默认情况下数据分布在多个分区，落到 HDFS 或本地就是一堆part-000xx文件。课设交报告时给老师一个 30 个碎片的目录体验很差，coalesce(1)把分区合并成一个，落盘后就是一个完整文件。代价是 shuffle 到单节点会损失并行度，数据量大了会 OOM。这份课设只有 5 万行，压力不大，放心用。

5. 避坑清单：爬取、类型与分区，五个常见翻车现场

5.1 接口返回空列表，导致城市数量远低于预期

现象：爬完所有城市后，passed_weather_ALL.csv行数接近 0 或者明显少于 5 万行，部分省份整片缺失。

原因：不是所有城市都有自动气象站。部分县级市和偏远地区的站点没有接入中央气象台的数据回传，passed/编码接口直接返回[]。代码只对「有数据」的城市计数并写入，所以这部分城市就直接蒸发了。

解决：在get_passed_weather()里遇到空列表时打印日志，把空数据城市的编码收集起来，爬完看一眼missing_cities.txt，能快速判断哪些区域数据缺失。如果缺失的比例超过 10%，建议换一个时间重新爬，而不是直接放弃——有些站点数据是暂缺，过几小时再请求就有了。

5.2 降雨量字段出现 9999，求和结果直接崩坏

现象：某城市 24 小时累计降雨量算出来是 9999 多毫米，明显违背常识，但数据明明是从官网接口拿的。

原因：气象数据里的 9999 通常代表「缺测」或「无观测值」，是行业惯例。接口里的rain1h字段没有用 null 表示缺失，而是用了一个不合理的巨大数值。如果照常数求和，这个城市会被顶上全国降雨量第一。

解决：报告里的做法是filter(col("rain1h") < 1000)，一杆子打死所有大于 1000 的异常记录。更稳妥的阈值可以参考：普通城市一小时降雨量超过 100 毫米已经属于极端暴雨，1000 这个阈值已经留了十倍余量，直接沿用没有任何问题。另外也可以在 cast 后先看agg(min, max)了解字段的取值区间再决定阈值。

5.3 CSV 用 Excel 打开中文乱码

现象：processed 好的passed_weather_ALL.csv用文本编辑器打开正常，拖进 Excel 后省份字段全变成「鐪佷唤」样式的乱码。

原因：Pandas 或 csv 模块在 Linux 下默认用 UTF-8 编码写文件，Excel 在无 BOM 的情况下会误判成 ANSI 或 GBK。报告里在写文件时用了codecs.BOM_UTF8，前缀加 BOM 后 Excel 能正确识别为 UTF-8。

解决：严格按照报告写法，open(file, 'ab+')后在开头写入一次 BOM 再写表头。如果你的代码已经跑完但没加 BOM，用下面的命令批量补：

sed -i '1s/^/\xef\xbb\xbf/' input/passed_weather_ALL.csv

注意这条命令只处理已有文件，新代码里还是要按报告里的方式写入。

5.4 日平均气温普遍偏低，个别城市直接缺失

现象：算出来的全国日平均气温 Top N 里，数值普遍比当天实际体感低 2-3 度，甚至有些城市直接没出现在结果里。

原因：问题几乎肯定出在「四时次不齐」。爬虫不是在同一秒拿到所有城市的数据，而是一个城市一个城市依次请求。如果你在当天 20 点出头开始爬，前几个城市能拿到 20 时的数据，排在队伍后面的城市在请求时 20 时数据还没回传，拿到的最晚时次是 14 时。cnt == 4过滤后这些城市被丢弃，而能凑齐四个时次的城市可能只有凌晨观测到完整链条的一部分，平均值自然偏低。

解决：把爬取时间放在当天 20 时 30 分之后，给站点数据回传留足缓冲。如果已经来不及重爬，可以把filter(col("cnt") == 4)放宽为cnt >= 2，并在结果表里加一个cnt列标注参与计算的时次数，这样图表下方的说明文字能给出解释空间。

5.5 collect() 爆内存或 writer 输出一堆碎文件

现象：调用collect()把 DataFrame 转成 Python list 用于可视化时，Jupyter 内核直接卡死；或者write.csv()落盘后目录里出现了几十个part-00000文件。

原因：collect()会把全部分区数据拉到 driver 端，课设数据只有 5 万行没问题，但如果没写coalesce(1)而默认分区数量大，driver 内存占用高到一定程度就会崩。碎片文件则是分区数量的锅，默认情况下 Spark 按 200 个分区计算（或者按 HDFS 块大小切），写盘时每个分区对应一个文件。

解决：可视化只取 TopN 用head(20)而不是collect()；如果确实要把全量结果转 list，先.coalesce(1)再.collect()。落盘统一走coalesce(1).write.csv。注意真集群里 coalesce(1) 会影响执行效率，但课设单机跑没毛病。

6. 可视化与结果验证：把 Spark 算出来的数字画成可信的图

6.1 matplotlib 中文与字体：不加这两行全是方块

跑完 Spark 拿到 TopN 和日平均气温数据后，下一步是用 matplotlib 画图。Ubuntu 下直接跑中文图例大概率显示乱码方块，需要先处理字体：

import matplotlib.pyplot as plt plt.rcParams['font.sans-serif'] = ['WenQuanYi Zen Hei', 'SimHei'] plt.rcParams['axes.unicode_minus'] = False # 解决负号显示方块 # 检查系统是否有可用的中文字体 import matplotlib.font_manager as fm fonts = [f.name for f in fm.fontManager.ttflist] print([f for f in fonts if 'CJK' in f or 'Hei' in f or 'Song' in f])

逻辑说明：第一行指定无衬线中文字体，第二行关掉 Unicode 负号。如果你没装任何中文字体，需要先执行sudo apt-get install fonts-wqy-zenhei安装文泉驿字体。axes.unicode_minus = False是很容易漏的一行——坐标轴上出现负温度时，matplotlib 默认的 minus 符号在部分字体下画成方块。

画降雨量 Top10 的柱状图：

# top20_rain 是从 Spark 里 head(20) 拿到的 Row 列表 cities = [row.city_name for row in top20_rain[:10]] rains = [row.rain24h for row in top20_rain[:10]] plt.figure(figsize=(10, 6)) plt.barh(cities[::-1], rains[::-1], color='#4C72B0') plt.xlabel('24h累计降雨量 (mm)') plt.ylabel('城市') plt.title('全国城市 24 小时累计降雨量 Top 10') plt.tight_layout() plt.savefig('output/rain_top10.png', dpi=150)

逻辑说明：barh是横向柱状图，城市名在纵轴，更容易阅读长文本。[::-1]把数据反转，让最大的值显示在图顶部。savefig前先tight_layout()防止长城市名被截断。到这里，Spark 算出来的数据就变成了可以直接贴进课程设计报告里的成品图。

6.2 结果一致性校验：Spark 输出与原始 CSV 手动对账

很多人跑完 Spark 就急着画图，我一般建议在画图前加一个快速校验步骤。方法很简单：从passed_weather_ALL.csv里挑一个排名靠前的城市，手工用 pandas 或者 Excel 算一下求和，和 Spark 输出的rain24h对比。

import pandas as pd raw = pd.read_csv('input/passed_weather_ALL.csv') # 挑一个 Top10 里的城市手工核对 city = top20_rain[0].city_name manual = raw[raw.city_name == city]['rain1h'].astype(float).sum() spark_val = top20_rain[0].rain24h print(f"手动计算: {manual:.2f} mm, Spark 结果: {spark_val:.2f} mm") assert abs(manual - spark_val) < 0.01, "结果不一致，检查数据"

逻辑说明：raw['rain1h'].astype(float).sum()是 pandas 里的暴力求和，原理是先把字符串列转数值再累加。Spark 端的结果来自 DataFrame 的agg(sum("rain1h"))。两者如果有出入，优先检查原始 CSV 里有没有空字符串或非数字字符——Spark 的cast("double")遇到脏字符会转成 null，而 pandas 的astype(float)会直接抛异常，这正是排查线索的差异。

从那以后，我每次跑完一个 Spark 分析任务，都会强制自己走一遍「原始文件 → 手工计算 → DataFrame 结果」三面对账。课设里的数据量小，用 pandas 验一条记录的成本几乎为零，但能挡住大部分类型转换和过滤条件的隐形 Bug。如果你在这个项目里卡在哪个环节，优先回看 CSV 原始文件，那里面藏着 80% 的答案。希望这篇拆解能帮你把这条路走得顺一点。

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