简介面向大数据课程期末课设与 Spark 入门实战的完整文档资料选题聚焦气象数据的采集、处理、分析与可视化。压缩包仅含 1 个 docx 文件大小约 744KB内容按项目背景、实验环境、数据来源、数据获取、数据分析、可视化与总结组织便于直接参考课程设计报告的结构与写法。文档以 Ubuntu 16.04、Python 3.9、Spark 2.4.0 和 Jupyter Notebook 为环境围绕中央气象台近 24 小时 2412 个城市、57888 条气温、降水、风力、气压和湿度数据展开涉及 JSON 接口抓取、缺失与异常处理、按城市聚合平均气温和降水量以及用 matplotlib 呈现分布和趋势。已有 13411 人学习下载适合需要 Spark 案例、课设模板和数据处理流程参考的学生。1. 数据到手先问一句2412 个城市的整点天气怎么用 Spark 一次算完拿到一份课设源码我第一件事不是看代码而是想它替我省了哪一段重复劳动。这份基于 Spark 的气象数据处理与分析就是这么个东西它把爬全国城市整点天气 → 落成 CSV → 用 Spark SQL 算 24 小时累计降雨量和日均气温 → 画图这条链路从头到尾走通了一遍2412 个城市、57888 条整点记录一台 Ubuntu 单机跑完不费劲。代码里用到的SparkSession、DataFrame.groupBy、agg、coalesce(1)都是 Spark SQL 的高频动作正好适合做大数据课设也适合想找一个端到端小项目练手的人。它不追求算法多新追求的是链路完整、每一步都有真实数据可验证——这对期末周赶着交东西的人来说比一堆漂亮架构图有用得多。2. 工程骨架与环境Spark 2.4、Python 3.9 这套组合怎么落地这个项目的目录结构非常简单没有 Maven、没有 Docker、没有调度框架就是一个脚本集合加几个数据目录。它符合课设的规模也符合新手第一次碰 Spark 时最该有的形态文件在哪儿、输出在哪儿一眼能对上。2.1 目录结构与版本清单我把它整理成下面这个样子方便直接照着建spark-weather/ ├── input/ # 爬虫产物也是分析入口 │ ├── province.csv # 省份列表 │ ├── city.csv # 城市列表含所属省份、站号 │ └── passed_weather_ALL.csv # 各城市最近 24 小时整点数据 ├── output/ │ ├── rain24h/ # 累计降雨量结果Spark 输出的 part 文件 │ └── avg_temperature/ # 日均气温结果 ├── crawler.py # 数据获取 ├── analysis_rain.py # 降雨量分析 ├── analysis_temperature.py # 气温分析 └── visualize.ipynb # Jupyter 里出图对应的环境版本原课设写得很明确组件版本说明操作系统Ubuntu 16.04单机模式不需要集群Python3.9跑 PySpark 和爬虫Spark2.4.0用local[*]就够数据量不到十万行Jupyter Notebook任意稳定版交互式调试和可视化matplotlib / tkinter系统包出图用可视化依赖装系统包最省事sudo apt-get install python3-matplotlib sudo apt-get install python3-tk这里有个前置条件容易被忽略Spark 2.4 是跑在 JVM 上的机器上得有 JDK 8。装完 JDK 再配 Spark 的环境变量顺序别反。sudo apt-get install openjdk-8-jdk tar -zxvf spark-2.4.0-bin-hadoop2.7.tgz -C /opt # 写进 ~/.bashrc开新终端才生效 export SPARK_HOME/opt/spark-2.4.0-bin-hadoop2.7 export PATH$SPARK_HOME/bin:$PATH export PYSPARK_PYTHONpython3 export PYSPARK_DRIVER_PYTHONpython3PYSPARK_PYTHON决定 executor 用哪个 Python 解释器PYSPARK_DRIVER_PYTHON决定 driver 用哪个。这俩变量看着像重复其实管的是两个进程只配一个的话在 Jupyter 里很容易出现driver 跑得通、executor 找不到模块的怪现象我踩过不止一次。2.2 Jupyter 里怎么把 Spark 接进来Jupyter 的 kernel 进程不会自动读SPARK_HOME直接from pyspark import SparkContext大概率报No module named pyspark。项目里的做法是在 notebook 开头先初始化import findspark # 参数换成你机器上 spark 的解压路径 findspark.init(/opt/spark-2.4.0-bin-hadoop2.7) from pyspark.sql import SparkSession spark (SparkSession.builder .appName(weather-notebook) .master(local[*]) # 用掉本机所有核 .config(spark.sql.shuffle.partitions, 8) # 小数据别用默认 200 .getOrCreate()) print(spark.version)findspark做的事就是把SPARK_HOME/python塞进sys.path顺便把pyspark依赖的 zip 包也加进去。它不能替代环境变量只是补了 notebook 这一层的缺口。最后的spark.sql.shuffle.partitions是我加的原课设没写。默认值 200 是给集群大数据集准备的这里只有几万行200 个 shuffle 分区意味着 200 次小任务调度跑起来反而更慢日志里一片Stage空转。2.3 三步冒烟测试确认环境是对的别一上来就跑全量分析先做最小验证。我一般的顺序是# 1. Java 和 Spark 都能起来 java -version spark-submit --version # 2. Python 能 import pyspark python3 -c import pyspark; print(pyspark.__version__) # 3. 真跑一个 RDD 作业 python3 -c from pyspark.sql import SparkSession s SparkSession.builder.master(local[2]).appName(smoke).getOrCreate() print(s.range(5).toPandas()) 第三步能打印出 0 到 4说明 driver、executor、序列化这一整条链路是通的。这一步跑不过后面所有AnalysisException都别急着往代码上找原因。3. 数据获取链路三层 JSON 接口串起来的爬虫这个项目在数据获取上比分析部分更有信息量。它没有解析 HTML而是直接找接口拿 JSON——这是抓数据最省事也最稳的路子。3.1 接口的三层结构打开站点控制台切省份、切城市能看到请求把数据分成了三级层级请求路径返回内容第一层/f/rest/province全部省份每个带名称和一个 URL第二层/f/rest/province/{省份编码}该省下所有城市含城市名和站号第三层/f/rest/passed/{城市站号}该城市最近 24 小时整点数据依赖关系是硬性的省份的三位编码要从第一层结果里拿城市站号要从第二层结果里拿所以只能串行地一层层往下抓。这也是为什么全量爬一次要半小时以上——2412 个城市每个城市一次请求加上节流时间就上去了。为了避免每次调试都重爬项目做了一个很实用的约定抓过就落盘下次直接读本地。BASE https://数据源域名 # 替换成实际站点域名 def province_url(): return f{BASE}/f/rest/province def city_url(pcode): return f{BASE}/f/rest/province/{pcode} def passed_url(city_code): return f{BASE}/f/rest/passed/{city_code}这三行看起来没内容但把 URL 拼装从业务逻辑里抽出来之后改数据源、加代理参数、加签名都只改这三行。原始代码把字符串硬编码在十几个地方改一次要全文替换很容易漏。3.2 把 Python 2 风格的爬虫改成能跑的版本原课设的爬虫源码有明显的 Python 2 痕迹print xxx这种语句式写法、sys.setdefaultencoding(utf8)、urllib3.Request。Python 3 下这几处会直接语法错误或者 AttributeError。我的处理是重写成urllib.request 重试 统一错误处理import json import time import urllib.request as ureq def get_json(url, retry3, timeout10): 请求一个 JSON 接口失败重试最终失败返回空列表而不是抛异常。 for i in range(retry): try: req ureq.Request(url, headers{User-Agent: Mozilla/5.0}) with ureq.urlopen(req, timeouttimeout) as resp: raw resp.read().decode(utf-8) # 显式指定编码别靠猜 return json.loads(raw) if raw else [] except Exception as e: print(f[warn] {url} 第 {i 1} 次失败: {e}) time.sleep(1 i) # 退避一下再试 return []几个参数值得说清楚。retry3是因为这类公开接口偶发 502 很常见重试一次往往就好了没必要让整个流程中断。timeout10必须设urlopen默认没有超时遇到不响应的连接会一直挂着全量跑的时候卡在某个城市上是很难排查的。返回空列表而不是抛异常是为了让上层能统一用if not data: continue处理——有些城市压根没有整点数据这不是错误是数据现状。还有一处原代码的write_csv每追加一次就往文件头写一个codecs.BOM_UTF8。分批落盘时写了十几次 BOM中间那些 BOM 会变成数据行里的乱码字符用 Spark 读进来字段名前面多个\ufeffselect(city_name)直接报列不存在。正确做法是只在文件首次创建时写一次 BOMimport csv, os def append_rows(path, rows): if not rows: return first not os.path.exists(path) or os.path.getsize(path) 0 # 首次用 utf-8-sig自动写 BOM之后用 utf-8 追加避免中途插 BOM enc utf-8-sig if first else utf-8 with open(path, a, newline, encodingenc) as f: writer csv.DictWriter(f, fieldnameslist(rows[0].keys())) if first: writer.writeheader() writer.writerows(rows)newline是 Windows 上防多余空行的必备参数Linux 上加上也不会有副作用我一般无脑加。3.3 分批落盘与断点续爬全量爬一次半小时中途断掉重来谁都受不了。项目里的做法是每 50 个城市落一次盘我在这里把边界条件补得更干净def crawl_all(cities, out_path, batch50): buf, done [], 0 for c in cities: data get_json(passed_url(c[code])) if not data: continue # 缺测城市直接跳过别写空行 done 1 for item in data: item.update( city_indexdone, city_codec[code], city_namec[city], provincec[province], ) buf.extend(data) if done % batch 0: append_rows(out_path, buf) print(f[info] 已落盘 {done} 个城市) buf [] time.sleep(1) # 节流别把对方接口打崩 append_rows(out_path, buf) # 收尾不足一批的残余done只在真有数据时才自增所以批次边界永远落在有数据的第 50、100、150 个城市上。原代码里count % 50 0的写表头判断依赖count恰好等于 50如果前 49 个城市全部缺测第一个批次就写不出表头后面 Spark 读的时候会把这批数据当成无表头文件。这种问题在小样本测试时很难发现全量跑才发现非常耽误时间。最后append_rows(out_path, buf)这句收尾不能省2412 不是 50 的整数倍最后一次不满 50 个的残余数据全靠它落盘。3.4 落盘后的字段结构最后产出的passed_weather_ALL.csv实际用到的字段就这几个字段含义分析里的用途province省份名称分组维度city_name城市名称分组维度 显示city_code城市站号分组维度唯一标识city_index自增计数只作标识不参与计算time整点时间拆出日期和小时temperature整点气温算日均气温rain1h过去 1 小时降水量累加成 24 小时雨量分组的时候province city_name city_code三个字段一起用是因为城市名会重名光靠中文名分组会把不同省份的同名城市合并。4. 核心分析模块拆解两段 Spark SQL 撑起全部分析分析部分只有两个指标但这两段代码把 Spark DataFrame 的常规操作覆盖得挺全select、cast、filter、groupBy、agg、sort、cache、coalesce、write。看懂这两段Spark SQL 的日常用法基本就通了一半。4.1 读取与清洗先让脏数据变成能算的数字原始 CSV 里所有字段都是字符串rain1h可能是None、空串也可能是 9999 这种缺测标记。直接sum会得到字符串拼接或者 null 传播所以第一步必须显式转型。from pyspark.sql import SparkSession from pyspark.sql.functions import col spark (SparkSession.builder .appName(weather-rain24h) .master(local[*]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate()) # inferSchema 关掉让所有列先以字符串进来转型交给后面显式做 df (spark.read .option(header, True) .option(encoding, UTF-8) .csv(input/passed_weather_ALL.csv)) df_rain (df.select(province, city_name, city_code, col(rain1h).cast(double).alias(rain1h)) .filter(col(rain1h).isNotNull()) .filter(col(rain1h) 1000))cast(double)遇到None这类非数字字符串返回 null所以后面必须补一句isNotNull()过滤否则聚合出来的求和结果全是 null——这个现象很迷惑人表能出来、列也有值全是空第一次遇到会以为是groupBy写错了。 1000这个阈值是原课设定的用来剔除明显超出物理范围的缺测标记。这里要说清楚降水量单位是毫米24 小时极端暴雨量级也就几百毫米单小时超过 1000 必然是异常值。阈值这东西最好写进配置而不是散在代码里后面讲到二次开发的时候还会提。4.2 24 小时累计降雨量分组求和 降序 单文件落盘口径很简单——按城市把 24 条小时降水量加起来。实现上就是groupByaggfrom pyspark.sql.functions import sum as _sum, round as _round df_rain_sum (df_rain .groupBy(province, city_name, city_code) .agg(_round(_sum(rain1h), 1).alias(rain24h)) .sort(col(rain24h).desc()) .cache()) # 后面要落盘 取 topN缓存避免重复计算 # coalesce(1) 把分区压到 1输出成一个 part 文件方便直接打开看 df_rain_sum.coalesce(1).write \ .mode(overwrite) \ .option(header, True) \ .csv(output/rain24h)逐个参数解释。_round(..., 1)保留一位小数气象数据没必要给 15 位浮点尾数。sort(...desc())是全局排序会触发一次全量 shuffle数据量小无所谓数据大了这步要挪到下游去。cache()是因为这条链路上被用了两次一次写入、一次取 topN不缓存的话整个分组求和会算两遍。coalesce(1)把输出压成一个文件——这一步在真实项目里要慎用所有数据挤到一个 task 上百万级以上会 OOM但这里只有几千条压成一个文件后直接双击就能看比面对一目录part-00000舒服得多。取 Top-N 给可视化用top10 df_rain_sum.limit(10).toPandas() # 只在结果集很小时用 toPandastoPandas()会把整个 DataFrame 拉回 driver 内存前面的limit(10)是关键漏了这句就是全量拉取几万行还行几十万行 driver 直接罢工。4.3 日均气温四时次筛选 计数兜底日均气温这块有个容易被忽略的口径问题。按气象观测规范日平均气温取 02、08、14、20 四个时次的平均值不是 24 个整点值简单平均。所以整个链路是先拆出日期和小时 → 筛出四个时次 → 按城市和日期分组 → 只有凑齐 4 条记录的才算数。from pyspark.sql.functions import (to_timestamp, hour, date_format, count, sum as _sum, round as _round, col) df_temp (df .select(province, city_name, city_code, col(temperature).cast(double).alias(temperature), to_timestamp(col(time), yyyy-MM-dd HH:mm).alias(ts)) .filter(col(temperature).isNotNull()) .withColumn(date, date_format(ts, yyyy-MM-dd)) .withColumn(hh, hour(ts))) df_4 df_temp.filter(col(hh).isin(2, 8, 14, 20)) df_avg (df_4 .groupBy(province, city_name, city_code, date) .agg(count(*).alias(cnt), _sum(temperature).alias(t_sum)) .filter(col(cnt) 4) # 缺时次的不算避免平均被拉偏 .withColumn(avg_temperature, _round(col(t_sum) / 4, 1)) .select(province, city_name, city_code, date, avg_temperature) .sort(col(avg_temperature).desc()) .cache())to_timestamp(col(time), yyyy-MM-dd HH:mm)这一步是必须的。直接把字符串丢给date_format在部分版本上会返回 null因为函数期望的是 timestamp 类型隐式转换并不可靠。显式指定格式串解析行为稳定得多。filter(col(cnt) 4)是这段代码里最重要的一句。爬虫是全量串行抓的2412 个城市跑下来时间跨度可能超过一小时排在前面的城市和排在后面的城市它们最近 24 小时的起点并不一致。如果不做计数校验某些城市只有 3 个时次的数据求和除 4 得到的平均值就偏了而且这种偏差在结果表里看不出来——数字是合法的就是错的。这一句相当于给日均气温加了个质量门槛宁可少几个城市也不出一批假数据。顺带提一个命名问题原课设把求和字段直接叫avg_temperature实际存的是四个时次的和后面还要除 4 才是平均值。这种命名和语义对不上的地方二次开发时看代码会非常难受我建议中间字段统一叫t_sum最终结果再叫avg_temperature别用一个名字贯穿两个含义。4.4 结果落盘与可视化衔接两段分析走完output/下就是两个单文件 CSV。可视化在 Jupyter 里做直接用结果 DataFrame 转 pandas 画图import matplotlib.pyplot as plt # 中文字体先确认系统里有哪个不然图上的城市名全是方块 plt.rcParams[font.sans-serif] [WenQuanYi Micro Hei, SimHei] plt.rcParams[axes.unicode_minus] False # 负号显示成方块的老问题 top df_rain_sum.limit(10).toPandas().iloc[::-1] # 倒序让最大值在顶部 fig, ax plt.subplots(figsize(8, 5)) ax.barh(top[city_name], top[rain24h]) ax.set_xlabel(24 小时累计降雨量 (mm)) ax.set_title(累计降雨量 Top10 城市) plt.tight_layout() plt.show()axes.unicode_minus False这行看着莫名其妙但没有它坐标轴上的负号会显示成一个小方框图上出现口0.5这种字样第一次见会以为是字体装坏了。5. 避坑与排查这份课设最容易翻车的五处下面这五条是我照这份源码在自己机器上复现时真实撞上的。每条按看到什么现象 → 为什么 → 怎么改写照着对号入座能省掉大半天。5.1 环境与运行时现象一No module named pyspark但spark-submit明明能跑。原因Jupyter kernel 用的是自己的 Python 解释器SPARK_HOME只影响 shell 里的命令不会自动注入到 kernel 的sys.path。解决notebook 第一格findspark.init(/opt/spark-2.4.0-bin-hadoop2.7)路径一定写绝对路径。还有一个隐蔽变体findspark装在了系统 Python 里但 kernel 用的是 conda 环境一样报错。判断方法是import sys; print(sys.executable)看清楚 kernel 到底是哪个 Python再往那个环境里装findspark和pyspark。现象二Python 3.9 环境下 PySpark 作业卡在Collect阶段日志里一堆序列化相关的报错。原因Spark 2.4 的 Python 支持范围比较老配 Python 3.9 属于越界使用worker 反序列化时可能撞上标准库变动。解决稳妥做法是给这个项目单独建一个 Python 3.6 或 3.7 的 conda 环境PYSPARK_PYTHON指向它driver 用哪个版本无所谓executor 必须对得上。环境变量和 notebook 里PYSPARK_PYTHON设成一致才不会出现driver 3.9、executor 3.7这种混搭。5.2 数据与口径现象三select(city_name)报cannot resolve但 CSV 表头明明写着city_name。原因分批写盘时每批都往文件里插了 BOM中间那些 BOM 变成了行首的不可见字符Spark 读到的列名是\ufeffcity_name。解决只在文件首次创建时用utf-8-sig后续追加一律utf-8见 3.2 的append_rows。已经生成的文件用sed或 Python 清一遍中间 BOM或者干脆删了重爬——数据在本地重爬比重洗便宜。现象四降雨量结果表有几万行但绝大多数rain24h是 null。原因cast(double)把None、--、空串都转成了 nullsum遇到全 null 的分组返回 null而分组本身仍然成立所以行数看着正常值全空。解决cast后立刻filter(col(rain1h).isNotNull())把无效记录挡在聚合之前。同时建议在爬虫落盘时就统一缺测值的表示别让None、空串、9999 三种写法同时存在。现象五不同城市算出来的24 小时累计雨量互相对不上排名前面的城市反而数值偏小。原因串行爬取耗时超过一小时各城市的 24 小时窗口起点不一致再叠加部分城市本身缺整点数据24 小时实际上变成了21 小时或19 小时。解决这是数据采集方式带来的口径问题没法靠改聚合逻辑修掉。可选的路有三条一是接受现状在报告里明确写清楚口径二是在 CSV 里补一列抓取时间戳分析时按窗口对齐三是改成并发抓取把总耗时压到几分钟内窗口自然就一致了。课设场景下第一条就够了但一定要写进文档不然答辩被问到会很难解释。6. 二次开发改造把它推到能讲清楚的三条线上这份课设能跑通但它更像一次性的实验脚本不是一个能长期维护的东西。要做二次开发我建议从三条线上动改动量都不大但收益明显。第一条线是参数化。现在文件路径、阈值、时次列表全硬编码在脚本里改个阈值要翻代码。抽一个配置字典出来脚本改成读参数# config.py CONF { input_path: input/passed_weather_ALL.csv, rain_max: 1000, # 缺测阈值按数据源实际情况调 hours: [2, 8, 14, 20], # 日平均气温的四个时次 partitions: 8, }然后analysis_temperature.py改成python analysis_temperature.py --config config.py这样跑。改完你会发现同一个脚本能同时跑四时次平均和24 小时平均两套口径对比起来很直观。第二条线是窗口对齐。给爬虫加一个crawl_ts字段每条记录写入抓取时刻分析时先按crawl_ts聚合成批次再在批次内算 24 小时指标。这样不同批次的结论不会混在一起也顺手解决了 5.2 里说的口径不一致——只不过这次是靠数据本身说清楚而不是靠文档说明。第三条线是加指标。现有两个指标太单薄气象数据里现成的可用字段还有风力、气压、相对湿度。加一个体感温度或者分区降雨分布都不难本质上还是groupBy加一个自定义表达式。但要提醒一句新增指标前先把口径写清楚比如最高气温到底是四时次里的最大值还是 24 个整点的最大值这两种算法在昼夜温差大的地区结果能差好几度。我见过太多人代码写对了、口径没定最后两版结果对不上回头查一整天。最后说个习惯。我做完这类分析项目第一件事不是画图而是把结果表按城市数、行数、null 比例三个数打一遍跟自己心里预期的量级对一下——2412 个城市结果表应该是两千多行而不是两万多行null 比例应该接近零。这三行检查代码不到十行但能拦住九成的翻车比事后翻日志快得多。从那以后我每次跑聚合都先看一眼这三个数。希望帮到你。本文还有配套的精品资源点击获取
