企业官网建设流程全解析

简介：面向大数据课程期末课设与 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 里出图

对应的环境版本，原课设写得很明确：

可视化依赖装系统包最省事：

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_PYTHON=python3 export PYSPARK_DRIVER_PYTHON=python3

PYSPARK_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 接口的三层结构

打开站点控制台，切省份、切城市，能看到请求把数据分成了三级：

依赖关系是硬性的：省份的三位编码要从第一层结果里拿，城市站号要从第二层结果里拿，所以只能串行地一层层往下抓。这也是为什么全量爬一次要半小时以上——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, retry=3, timeout=10): """请求一个 JSON 接口，失败重试，最终失败返回空列表而不是抛异常。""" for i in range(retry): try: req = ureq.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with ureq.urlopen(req, timeout=timeout) 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 []

几个参数值得说清楚。retry=3是因为这类公开接口偶发 502 很常见，重试一次往往就好了，没必要让整个流程中断。timeout=10必须设，urlopen默认没有超时，遇到不响应的连接会一直挂着，全量跑的时候卡在某个城市上是很难排查的。返回空列表而不是抛异常，是为了让上层能统一用if not data: continue处理——有些城市压根没有整点数据，这不是错误，是数据现状。

还有一处：原代码的write_csv每追加一次就往文件头写一个codecs.BOM_UTF8。分批落盘时写了十几次 BOM，中间那些 BOM 会变成数据行里的乱码字符，用 Spark 读进来字段名前面多个\ufeff，select("city_name")直接报列不存在。正确做法是只在文件首次创建时写一次 BOM：

import 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="", encoding=enc) as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) if first: writer.writeheader() writer.writerows(rows)

newline=""是 Windows 上防多余空行的必备参数，Linux 上加上也不会有副作用，我一般无脑加。

3.3 分批落盘与断点续爬

全量爬一次半小时，中途断掉重来谁都受不了。项目里的做法是每 50 个城市落一次盘，我在这里把边界条件补得更干净：

def crawl_all(cities, out_path, batch=50): 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_index=done, city_code=c["code"], city_name=c["city"], province=c["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三个字段一起用，是因为城市名会重名，光靠中文名分组会把不同省份的同名城市合并。

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 条小时降水量加起来。实现上就是groupBy+agg：

from 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() # 只在结果集很小时用 toPandas

toPandas()会把整个 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"、"--"、空串都转成了 null，sum遇到全 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 比例应该接近零。这三行检查代码不到十行，但能拦住九成的翻车，比事后翻日志快得多。从那以后我每次跑聚合都先看一眼这三个数。

希望帮到你。

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