企业官网建设流程全解析

简介：这份面向大数据专业期末课设的Spark气象数据分析文档，适合需要完成课程设计、案例复现或入门Spark数据处理的学生参考。内容围绕气象数据获取、清洗、聚合分析与可视化展开，涉及Linux Ubuntu 16.04、Python 3.9、Spark 2.4.0及Jupyter Notebook环境，并包含中央气象台城市天气数据的爬取思路、JSON接口层级、CSV存储与缺失异常处理等实践环节。压缩包仅含1个docx，约744KB，为完整课程设计报告文档，目录覆盖项目背景、实验环境、数据来源、数据获取、数据分析、数据可视化与总结，便于按章节梳理实现流程和报告写法。已有13411人学习。读者可从中了解2412个城市、57888条气象数据的处理规模，以及平均气温和降水量计算、可视化呈现等关键任务，获得可参照的课设结构与排错思路。

1. 选题落地：为什么“基于 Spark 的气象数据处理与分析”适合当大数据期末课设

“基于 Spark 的气象数据处理与分析”这个题目在大数据期末课设里出现频率很高，但交上来的东西差别极大：有人给的是能跑通的完整链路，有人只交了张 groupBy 截图。它合适，是因为数据源公开、体量刚好卡在一个尴尬位置——2400 多个城市、5.7 万余条整点记录，pandas 也扛得住，但手写循环统计会明显变慢，正好把 Spark SQL 的分组聚合、类型转换、缓存、分区写出这几件事都逼着走一遍。这套方案要解决的是从零把数据抓下来、整理成规整宽表、用 Spark 算出各城市过去 24 小时累计雨量和当日四时次平均气温，再出图。适合要交课程设计、需要一份能照着搭结构的人。

2. 数据获取链路：省—市—逐小时三级接口的串接与分批落盘

2.1 接口结构：三层请求的依赖关系

这个项目最容易劝退人的地方不是 Spark，而是数据从哪来。打开目标气象服务站点的任意一个城市页面，按 F12 看 Network 面板，切换省份和城市，会看到页面主体其实是空的，数据全靠 XHR 异步拉回。翻几条请求就能归纳出规律：

关键在于第二层和第三层的参数不是写死的，得从上一层的结果里抠出来。省份的url字段长这样.../province/ABC.html，取文件名去后缀就得到三位编码；城市的code是五位编号，直接拼到/f/rest/passed/后面。所以爬取顺序是固定的：先省、再市、最后逐小时，跳步就会拿不到参数。

这条路子的好处是省掉了 HTML 解析。返回体本身就是 JSON，json.loads之后就是 list[dict]，不需要 BeautifulSoup，也不需要处理各种标签嵌套和反爬的 DOM 混淆。缺点也很明显：三层串行，任何一层返回空都会让后面全断，过滤逻辑必须写扎实。

2.2 爬虫骨架：分批刷盘与断点续爬

原始报告里的爬虫是 Python 2 写法（urllib2、print语句），直接拷到 Python 3 会全线报错，这里已经换成urllib.request。还有一个工程上的硬需求：全量爬一次超过半小时，中途断网或者手滑 Ctrl+C，重跑就得从头开始，所以必须做文件级断点。

import csv, json, os, time from urllib.request import urlopen, Request HOST = "http://<气象服务站点域名>" OUT = "input/passed_weather_ALL.csv" BATCH = 50 # 每凑满 50 个城市刷一次盘，防止内存堆积和意外丢数据 SLEEP = 1 # 城市之间停 1 秒，控制请求节奏 def get_json(path): """统一封装请求：拼 host、带 UA、超时兜底，空响应返回空列表""" req = Request(HOST + path, headers={"User-Agent": "Mozilla/5.0"}) with urlopen(req, timeout=10) as resp: body = resp.read().decode("utf-8").strip() return json.loads(body) if body else [] def flush(rows, path, write_header=False): """把缓冲区写成 csv；已存在就追加，第一次写才落表头""" if not rows: return mode = "w" if (write_header or not os.path.exists(path)) else "a" with open(path, mode, encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) if mode == "w": writer.writeheader() writer.writerows(rows) def crawl_all(): if os.path.exists(OUT): print("已存在数据文件，跳过爬取，避免重复劳动") return cities = [] for p in get_json("/f/rest/province"): code = p["url"].split("/")[-1].split(".")[0] # 从 url 尾部抠三位编码 cities.extend(get_json("/f/rest/province/" + code)) buf, n = [], 0 for city in cities: data = get_json("/f/rest/passed/" + city["code"]) for item in data or []: item.update(province=city["province"], city_name=city["city"], city_code=city["code"]) buf.append(item) n += 1 if data: pass if n % BATCH == 0: flush(buf, OUT, write_header=(n == BATCH)) buf = [] time.sleep(SLEEP) flush(buf, OUT) # 收尾：不足一批的尾巴也要落盘

几个参数的含义值得说清楚。BATCH控制的是内存水位和丢失窗口的平衡点：设成 50，最坏情况丢 49 个城市的数据，重爬成本可以接受；设成 500，内存占用上去、丢一次损失太大。SLEEP是礼貌间隔，虽然接口本身没做严格的频控，但把间隔压到 0 容易触发连接重置，反而要在异常处理上花更多时间。encoding="utf-8-sig"是为了让后续 Excel 打开不乱码，代价是多出 BOM 头，Spark 读的时候要留意这一点。

还有一个细节：city_index字段在原始报告里是用来计数的，但它的值取决于爬取当次的遍历顺序，重跑一次就可能对不上，所以真正做分组聚合时应该用city_code而不是这个序号。

2.3 数据字典与异常值预判

落盘后的宽表大概长这样，实际分析只用到其中一部分字段：

这里面有两个必须提前想好的处理。一是rain1h的异常值，报告里明确说了大于 1000 属于异常，正常一小时雨量不可能到这个量级，通常是接口把缺测或者占位编码直接返回了，所以计算前必须过一遍filter，否则某个城市的累计雨量会被拉到一个荒谬的数，榜单直接失去意义。二是temperature的缺测，通常的做法是先cast成 double，再看分布，如果出现明显偏离物理区间的值（比如绝对值远超常见气温范围），在分组求平均前先剔除，或者在四时次校验那一步自然淘汰掉。

另外还有一类隐性缺失：部分城市可能压根没有最近 24 小时数据，接口返回空数组。这类城市在爬取阶段就被if data过滤掉了，不会进 CSV，所以最终行数会小于「城市数 × 24」的理论值。这个差额在写报告时要主动说明，别让评审以为是丢数据。

3. 环境与技术选型：为什么落到 Spark SQL 而不是手写循环

3.1 环境清单与依赖安装

环境这套配置是课程设计里比较典型的组合，四个组件各管一段：

装完主体之后，可视化还差两个系统级包，缺了会在import matplotlib.pyplot或者弹窗渲染时直接报错：

# matplotlib 依赖 tkinter 做后端渲染，两个包要一起装 sudo apt-get update sudo apt-get install -y python3-matplotlib python3-tk

这里有个版本匹配的坑值得提前说。Spark 2.4.0 发布时官方主要覆盖到 Python 3.7 上下，配 3.9 属于「能跑但不保证」的组合。多数情况下 driver 端没问题，真正容易翻车的是 executor 端调 Python worker 的时候，报的错还很隐晦，看起来像序列化问题。我一般会先确认spark-submit或 Notebook 里PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON指向同一个解释器路径，两边不一致是最常见的诱因。如果实在绕不过去，把 Python 降到 3.7/3.8 往往比改代码更快。

3.2 选型对比：DataFrame API 与另外两条路

把这个需求拆开，本质就是「按城市分组求和」和「按城市+日期分组求平均」两个聚合。三条实现路径的差别不小：

选 DataFrame 的理由不是「时髦」，而是这套数据本来就是规整的二维表，字段固定、类型明确，用groupBy().agg()一句话能表达清楚的事情，没必要退回到 RDD 层面自己写键值对。而且explain()能看到 Catalyst 生成的物理计划，做投影下推和聚合优化，这部分不用自己操心。课程设计里能把select → filter → groupBy → agg → sort这条链讲明白，比堆一堆mapPartitions更能体现对引擎的理解。

3.3 CSV 读取的类型陷阱

spark.read.csv有个默认行为：所有列都按 string 读。这意味着哪怕 CSV 里写的是23.5，读进来也是字符串，直接sum会得到 null 或者报错。有两种处理方式，但只有一种适合这个场景。

from pyspark.sql import SparkSession from pyspark.sql.functions import col spark = (SparkSession.builder .appName("weather-analysis") .master("local[*]") # 课设环境用本地多线程即可 .getOrCreate()) df = (spark.read .option("header", True) .option("encoding", "UTF-8") # 配合写入时的 utf-8-sig .csv("input/passed_weather_ALL.csv")) # 显式指定类型，不用 inferSchema df_typed = (df.withColumn("temperature", col("temperature").cast("double")) .withColumn("rain1h", col("rain1h").cast("double")))

inferSchema=True看起来省事，但它会额外触发一次全表扫描来推断类型，数据量一大就慢一倍；更要命的是推断结果不稳定——同一列里混进几个空值或者异常字符串，整列可能被判成 string，后面照样要手动 cast。所以这里的做法是显式 cast，把类型控制权留在自己手里。

UTF-8那个选项也值得提一句。写入时用了utf-8-sig会带 BOM，如果读取时不指定编码，某些环境下列名第一个字段会带上不可见字符，导致col("province")取不到值，报的错还是「列不存在」，排查起来很费时间。加上编码选项就能绕过去，或者干脆写文件时用普通utf-8。

4. 分析算子设计：累计雨量与四时次日均温两条主线

4.1 累计降雨量：分组求和、异常过滤与结果落盘

第一条主线是算每个城市过去 24 小时的累计雨量。逻辑上就是按城市分组对rain1h求和，但要落地成正确的算子链，有几个位置不能省。

from pyspark.sql.functions import col, sum as _sum df_rain = (df_typed .select("province", "city_name", "city_code", "rain1h") .filter(col("rain1h").isNotNull() & (col("rain1h") < 1000))) # 剔除异常 df_rain_sum = (df_rain .groupBy("province", "city_name", "city_code") .agg(_sum("rain1h").alias("rain24h")) .orderBy(col("rain24h").desc()) .cache()) # 后面既要写出又要取 Top-N，先缓存 # 结果落盘：合并成单文件方便查看，数据量小才敢这么干 df_rain_sum.coalesce(1).write.mode("overwrite").option("header", True) \ .csv("output/rain24h") # 取前若干条给可视化用，避免重复计算 top_rain = df_rain_sum.limit(20).collect()

filter里那个< 1000是这套方案能不能出正确结论的分水岭。缺了它，某个城市可能因为一条占位记录，累计雨量直接飙到四位数，稳居榜首，而实际它可能一滴雨都没下。isNotNull()也一样，cast失败会产生 null，不清掉的话sum会把整个分组的和变成 null，最终这个城市直接从榜单里消失，属于静默错误，比报错更难发现。

coalesce(1)在这个场景下是合理的：结果只有两千多行，合并成一个 part 文件方便后续读取和贴图；如果结果上百万行还这么写，就会把所有数据压到一个 executor 上，内存直接爆掉。cache()的位置也有讲究——放在orderBy之后，意味着缓存的是排好序的最终结果，后面write和limit().collect()都复用它；如果放在groupBy之前，缓存的还是原始明细，等于白占内存。

4.2 四时次日均气温：时间字段拆解与计数校验

第二条主线稍微绕一点，因为它不是简单求平均。按气象观测的通行口径，日平均气温取 02 时、08 时、14 时、20 时四个时次的数据求平均，这意味着要先从time里把日期和小时拆出来，筛出这四个时次，再做分组平均。

from pyspark.sql.functions import date_format, hour, count as _count, round as _round df_temp = (df_typed .select("province", "city_name", "city_code", "time", "temperature") .withColumn("temperature", col("temperature").cast("double")) .withColumn("date", date_format(col("time"), "yyyy-MM-dd")) # 取出日期 .withColumn("hour", hour(col("time")))) # 取出小时 df_4point = df_temp.filter(col("hour").isin(2, 8, 14, 20)) # 只要四个时次 df_4point = df_4point.filter(col("temperature").isNotNull()) # 气温缺测的先清掉 df_avg_temp = (df_4point .groupBy("province", "city_name", "city_code", "date") .agg(_count("temperature").alias("cnt"), _sum("temperature").alias("t_sum")) .filter(col("cnt") == 4) # 必须凑满 4 个时次 .withColumn("avg_temperature", _round(col("t_sum") / 4, 1)) .select("province", "city_name", "city_code", "date", "avg_temperature") .orderBy(col("avg_temperature").desc()) .cache()) df_avg_temp.coalesce(1).write.mode("overwrite").json("output/avg_temperature") avg_list = df_avg_temp.collect() # 转成 list 交给 matplotlib

filter(cnt == 4)这一步是整个算子链的核心校验，也是最容易被省掉的一步。原因很实在：24 个整点里只要丢了 08 时那一条，这个城市当天的「四时次平均」就变成「三时次平均」，数值可能偏高也可能偏低，而且从结果上完全看不出来。加上计数校验之后，凑不满四次的直接淘汰，宁可少几个城市，也不能让口径不一致的数据混进同一张榜单。

另外原始报告里把求和结果直接命名成avg_temperature，这一步其实少了个除法。_sum出来的是四个时次的和，必须再除以 4 才是平均值，round(..., 1)保留一位小数是为了和气象播报的习惯对齐。这种「字段名和实际含义对不上」的写法在读代码时特别容易误解，变量名最好和真实语义一致，t_sum和avg_temperature分开命名。

4.3 结果出图：从 DataFrame 到可读的图表

数据分析做完了不出图，答辩的时候说服力会弱一截。collect()把结果拉回 driver 变成普通 list 之后，剩下的就是 matplotlib 的活。

import matplotlib matplotlib.use("Agg") # 无图形界面环境下必须用非交互后端 import matplotlib.pyplot as plt # 中文字体：Ubuntu 上通常没有 SimHei，用文泉驿或思源黑体 plt.rcParams["font.sans-serif"] = ["WenQuanYi Zen Hei", "Noto Sans CJK SC", "SimHei"] plt.rcParams["axes.unicode_minus"] = False # 负号正常显示 top = df_rain_sum.limit(15).collect() # 反转顺序，让最大值出现在条形图顶部 names = [r["city_name"] for r in top][::-1] values = [r["rain24h"] for r in top][::-1] fig, ax = plt.subplots(figsize=(8, 6)) ax.barh(names, values, color="#4C72B0") ax.set_xlabel("过去 24 小时累计雨量 (mm)") ax.set_title("累计雨量 Top-15 城市") ax.grid(axis="x", linestyle="--", alpha=0.4) fig.tight_layout() fig.savefig("output/rain_top15.png", dpi=150)

中文乱码是这条路上必踩的一脚，现象是图上所有汉字变成一排方块，报错信息里还带着missing from font的警告。根因不是 matplotlib 坏了，而是默认字体族里没有中文字形。Ubuntu 上用fc-list :lang=zh查一下装了哪些中文字体，把名字填进font.sans-serif数组最前面就行。axes.unicode_minus = False是配套的另一个小坑，不关掉的话负号也会变成方块，气温图里一旦出现零下数值就会露馅。

气温那条线适合画分布而不是榜单：两千多个城市的日均温，用直方图看整体分布形态，再用横向条形图对比几个代表性城市的差异，比堆一张密密麻麻的折线图清楚得多。

4.4 缓存、分区与写出的取舍

coalesce和cache这两个算子在小数据量下容易被当成装饰品随手加，实际上各有各的适用边界。

cache()只在「同一个 DataFrame 被多次行动操作触发」时才划算。这个项目里两条主线的结果各被用了两次（一次写出、一次 collect 画图），所以缓存有意义；如果只是算完就写盘、后面不再用，缓存反而是纯开销，还占内存。用完想主动释放，调unpersist()。

coalesce(1)的取舍更直接：结果小、需要单文件交付，就用；数据量大、或者这是中间结果，就别碰。判断标准很简单——结果集能不能舒适地放进单个 executor 的内存。两千多行的聚合结果是能，原始 5.7 万行明细就不是必需合并，让它按分区写多个 part 文件反而更稳。

5. 避坑清单：五个最容易翻车的位置和排查顺序

5.1 爬到一半断了，重跑从头开始

现象：终端里跑了几百行城市名，突然一个网络超时抛出异常，进程退出，input/目录下只有一个残缺的 CSV，或者什么都没有。

原因：缓冲区全在内存里，没有中途刷盘；同时也没有文件存在性检查，重跑时不会跳过已完成的城市，只能整体重来。

解决：按批次flush到磁盘，并且把「文件存在即跳过」写进入口函数。真要做到城市级断点，还可以给每个城市单独落一个小文件，或者在 CSV 里记一个已完成编码集合。课程设计里做到批次级就够了，但心里要清楚这只是降低损失，不是真正幂等。

5.2 各城市时间窗口对不齐，累计雨量口径不一致

现象：算出来的累计雨量排名看着挺正常，但换个时间点重爬一次，结果差异很大，某些城市的名次能跳几十位。

原因：三层接口是逐个城市串行请求的，两千多个城市爬完要半小时以上，第一个城市拿到的是「当时往前 24 小时」，最后一个城市拿到的是「半小时之后往前 24 小时」。每个城市的窗口本身没错，但整批数据不在同一个时间基准上。

解决：在结果里带上实际数据的起止时间，报告里明确写出「各城市时间窗口存在分钟级偏移」这个前提；或者爬完之后按time字段筛出交集时段再聚合。多数课设接受前者，但必须写清楚，别把口径差异当成分析结论。

5.3 异常值没清干净，均值被一条记录带飞

现象：某城市的累计雨量数值明显不合常理，或者日均气温算出来偏离区间老远。

原因：接口对缺测数据返回占位值或者极大值，cast成 double 之后语法上合法，聚合时照单全收。

解决：在聚合前加一道过滤，rain1h < 1000这类物理约束是硬性的；气温也建议加范围校验或者分布检查。过滤前后各count()一次，把被剔除的条数记下来，写进报告的「数据清洗」小节，比一句「做了清洗」有说服力得多。

5.4 四时次没凑齐也参与平均

现象：日均温表里出现一些城市，但复查原始数据发现它当天只有一个或两个时次有值。

原因：groupBy之后直接avg，Spark 会忽略 null 求平均，缺时次的城市照样出结果，只是口径悄悄变了。

解决：agg里同时带上count，再filter(cnt == 4)。这一步最好做成固定套路，凡是「按 N 个样本求平均」的场景都带上计数校验，别相信数据是齐的。

5.5 图表中文全是方块，或者图根本存不下来

现象：savefig之后打开图片，标题和坐标轴标签是一排方块；或者在某些环境里直接抛RuntimeError: main thread is not in main loop。

原因：前者是字体族里没有中文字形；后者是在无桌面环境里用了交互式后端。

解决：脚本开头matplotlib.use("Agg")强制非交互后端，字体用fc-list :lang=zh查到的实际名字填进去，同时关掉axes.unicode_minus。这三件事我一般写成一段固定的模板代码，每个新脚本开头直接贴，省得每次重新查。

6. 验收与答辩：怎么把一份课设讲成一条闭环

课程设计的分数不只看代码跑没跑通，更多看你能不能把「数据从哪来、口径怎么定、结论边界在哪」讲成一件事。我一般会按下面这张清单自查一遍再上台：

答辩现场被问得最多的三个问题基本固定：为什么用 Spark 不用 pandas、数据有缺失怎么办、你的结论能说明什么。第一个答选型对比（数据量、算子表达力、后续可扩展），第二个答具体处理策略加上被剔除的条数，第三个最考验人——两千多个城市一天的快照，能说明的是「这一时段内哪些城市累计雨量靠前、哪些城市日均温偏高」，说明不了气候趋势，也说明不了因果。把边界主动说出来，比硬拗一个有洞的结论得分高。

有个具体技巧分享：我会单独做一页「数据口径说明」，把时间基准、四时次定义、异常值阈值三件事列成三行，答辩时先放这一页再放结果。血泪经验是，评审最容易卡住的不是算法复杂度，而是「你这 24 小时到底是从几点算到几点」。这一页放出去，后面所有图表都不用再解释前提。

从那以后我每次做带时间窗的数据分析，都会在写第一行聚合代码之前先把口径写进 README，宁可多花十分钟，也不要在答辩台上被问住。希望帮到你。

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