简介面向大数据期末课设的一份气象数据处理与分析项目完整呈现从中央气象台公开网页接口获取数据、清洗处理、聚合统计到可视化的完整流程。实验围绕Spark分布式计算框架展开运行于Linux Ubuntu系统使用Python脚本与Jupyter Notebook环境覆盖全国两千余座城市、近六万条气温和降水记录并针对缺失或异常数据做了过滤。报告按项目背景、实验环境、数据来源、数据获取、数据分析、数据可视化与总结等章节展开附录中保留爬虫源码及关键实现思路适合大数据专业学生完成课程设计或复习实战流程。资源共一个文件为docx格式完整课程设计报告压缩包大小744KB已有超过一万三千人学习下载。借助报告可掌握气象接口请求解析、Spark DataFrame分组聚合、平均气温与降水量计算以及图表可视化等实用技能。1. 让 Spark 2.4.0 在 Python 3.9 下跑起来是这个课设的第一道坎做大数据期末课设的人十有八九会卡在第一步环境装好了from pyspark.sql import SparkSession一执行Jupyter 内核直接挂掉报一堆py4j或 socket 错误。这套基于 Spark 2.4.0 的气象数据处理与分析项目要求的环境组合是 Ubuntu 16.04、Python 3.9、Spark 2.4.0、Jupyter Notebook。表面看是读 CSV、分组、聚合这些 Spark 入门操作实际上坑全埋在版本匹配上——Spark 2.4.0 官方对 Python 3.9 的支持并不完整JDK 版本不对也会闪退。这篇文章按我自己复现这套课设的顺序写环境怎么选版本、爬虫怎么写、Spark SQL 怎么算 24 小时累计降水量和四时次平均气温、装坏了怎么救、最后怎么验证跑通。适合正在做 spark 实战项目、spark 数据处理课设的同学直接照着抄。2. 搭建实验环境Ubuntu 16.04、Python 3.9、Spark 2.4.0、Jupyter 的版本匹配2.1 先装 JDK 8不要直接装默认 OpenJDKSpark 2.4.0 是基于 JDK 8 编译和测试的如果你装的是 apt 默认的 OpenJDK 11 或更高版本Spark 启动时日志会刷大量的illegal reflective access警告到了 JDK 17 甚至可能连 worker 都起不来。我一般会在装 Spark 之前先用java -version确认当前版本java -version # 如果输出版本不是 1.8 开头先安装 OpenJDK 8 sudo apt-get update sudo apt-get install -y openjdk-8-jdk我的做法是装完以后把JAVA_HOME显式写进/etc/environment不要依赖 system 默认的 java 路径因为后续 Spark 脚本会优先读这个环境变量echo export JAVA_HOME/usr/lib/jvm/java-8-openjdk-amd64 /etc/environment echo export PATH$JAVA_HOME/bin:$PATH /etc/environment source /etc/environment这里是关键如果不设置JAVA_HOMESpark 的start-all.sh和pyspark会尝试用which java去找如果系统里有多个 JDK 或者软链接混乱会直接报Unable to locate a Java Runtime。另外要注意 Ubuntu 16.04 的 apt 默认源里 OpenJDK 8 是可以直接装的如果提示找不到包先sudo apt-get update刷新索引。2.2 下载并解压 Spark 2.4.0 二进制包Spark 的环境选择有个取舍点原项目数据规模是 57888 条完全属于单机 local 模式能处理的量级没必要装 Hadoop 集群。所以我选的是spark-2.4.0-bin-hadoop2.7.tgz这个包名字里带hadoop2.7只是表示它内置了与 Hadoop 2.7 配套的编译依赖在 local 模式下不需要单独部署 HDFS。# 下载 2.4.0 版本约 240MB wget https://archive.apache.org/dist/spark/spark-2.4.0/spark-2.4.0-bin-hadoop2.7.tgz # 解压到 /opt 目录并建立软链接方便后续切换版本 sudo tar -zxvf spark-2.4.0-bin-hadoop2.7.tgz -C /opt/ sudo ln -s /opt/spark-2.4.0-bin-hadoop2.7 /opt/spark # 添加环境变量 echo export SPARK_HOME/opt/spark /etc/environment echo export PATH$SPARK_HOME/bin:$PATH /etc/environment解压完成后要改一个配置文件路径是$SPARK_HOME/conf/spark-env.sh。第一次跑 pyspark 时如果没配SPARK_LOCAL_IP很可能会遇到 worker 之间互相连不上、Jupyter 里显示 SparkContext 一直处于 connecting 状态的问题。我的配置如下cp $SPARK_HOME/conf/spark-env.sh.template $SPARK_HOME/conf/spark-env.sh cat $SPARK_HOME/conf/spark-env.sh EOF export JAVA_HOME/usr/lib/jvm/java-8-openjdk-amd64 export SPARK_LOCAL_IP127.0.0.1 export PYSPARK_PYTHON/usr/bin/python3.9 EOF这里的SPARK_LOCAL_IP127.0.0.1是必须的因为在某些网络环境下 Spark 会尝试获取主机名对应的 IP如果/etc/hosts里主机名解析到 IPv6 或一个不可达地址本地模式下 SparkContext 会卡在注册阶段。2.3 Python 3.9 与可视化组件 matplotlib、tk 的安装Ubuntu 16.04 默认的 Python 是 3.5但项目要求 3.9所以这个步骤不能省也不能跳。我自己更推荐用源码编译方式装 Python 3.9虽然编译时间长一点但能保证和 Spark 的二进制兼容。如果嫌麻烦也可以用deadsnakes第三方源在课程设计场景下也没有问题。# 编译依赖 sudo apt-get install -y build-essential libssl-dev zlib1g-dev libffi-dev # 下载 Python 3.9 源码并编译 wget https://www.python.org/ftp/python/3.9.13/Python-3.9.13.tgz tar -xzf Python-3.9.13.tgz cd Python-3.9.13 ./configure --enable-optimizations make -j4 sudo make install编译安装完成后注意确认python3.9能够直接执行并且pip3.9可用python3.9 -m pip install --upgrade pip python3.9 -m pip install jupyter matplotlib numpy pandas然后安装可视化需要的系统级组件。原项目里提到了两个 apt 包我复现时确认过它们是配套关系python3-matplotlib提供绘图库python3-tk提供 TkInter 后端如果只装前者在 Jupyter 里执行plt.show()时会报ModuleNotFoundError: No module named tkinter。sudo apt-get install -y python3-matplotlib python3-tk让我强调一下安装顺序先装 Python 3.9再装 pip 依赖最后装 apt 包。如果反过来了python3-matplotlib会安装到系统 Python 3.5 上而你在 Jupyter 里用的是 3.9 的内核照样 import 不到。2.4 Jupyter 集成用 PYSPARK_DRIVER_PYTHON 启动 NotebookSpark 和 Jupyter 的集成是这里最容易翻车的地方。有两条路一条是在 Jupyter 里手动创建SparkSession另一条是用pyspark命令自动拉起 Jupyter。课程设计项目我推荐第二条因为PYSPARK_DRIVER_PYTHON会替你把 SparkContext 和本机的 driver 网络都初始化好少踩很多坑。export PYSPARK_PYTHON/usr/bin/python3.9 export PYSPARK_DRIVER_PYTHONjupyter export PYSPARK_DRIVER_PYTHON_OPTSnotebook --no-browser --ip127.0.0.1 --port8888 # 启动 pysparkPYSPARK_PYTHON指给 driver 和 executor 要用的 Python 解释器PYSPARK_DRIVER_PYTHON则是把 driver 的入口程序替换成 Jupyter 的启动器。如果不设置PYSPARK_DRIVER_PYTHON_OPTSpyspark 只会打开一个空白 notebook 页面无法指定端口和访问地址。配置完成后在 Jupyter 里执行下面这段验证代码from pyspark.sql import SparkSession spark SparkSession.builder \ .master(local[2]) \ .appName(weather-analyse) \ .getOrCreate() print(spark.version)如果正常输出了2.4.0说明 pyspark 和 Jupyter 的集成成功了。这里local[2]是申请 2 个 CPU 核心跑本地任务57888 条气象数据用它绰绰有余。3. 数据获取把气象站点的 JSON 接口串成一条爬取流水线3.1 三个接口的层级关系原始实验数据从 nmc.cn 站点抓取数据量是 2412 个城市、57888 条记录。我在做这个任务时第一步不是写爬虫而是打开浏览器控制台通过切换省份、城市观察网络请求很快就能发现气象数据接口是三层嵌套的 JSON 异步接口请求 URL返回内容作用http://www.nmc.cn/f/rest/province省份列表每个省份带一个 URL 字段拿到省份编码http://www.nmc.cn/f/rest/province/{省份编码}该省下所有城市带城市编号拿到城市编号http://www.nmc.cn/f/rest/passed/{城市编号}该城市过去 24 小时整点天气最终天气数据所以必须先爬省份再爬城市最后爬天气。三个接口返回的都是 JSON不需要像爬 HTML 那样解析 DOM直接用json.loads即可。有个细节省份数据的url字段是一个相对路径比如某个省份的编码是ABJ取的时候要先用split(/)[-1].split(.)[0]把真实的省份编码提取出来再拼接到第二个接口上。# 提取省份编码的常见做法 province[url].split(/)[-1].split(.)[0]3.2 用 urllib 写一个可复跑的爬虫这里有一个重要的修正。原项目代码写的是urllib3.Request(url)但在 Python 3 环境下urllib3并不提供Request类正确写法是urllib.request.Request。如果不改直接跑会报AttributeError: module urllib3 has no attribute Request。下面是我实际跑通的版本# 导入模块 import json import csv import codecs import os import time import urllib.request class WeatherCrawler: def get_json(self, url): req urllib.request.Request(url) req.add_header(User-Agent, Mozilla/5.0 (X11; Linux x86_64) Chrome/108.0) with urllib.request.urlopen(req, timeout10) as resp: body resp.read() return json.loads(body) def write_csv(self, file, data): print(begin to write to file) with open(file, a, encodingutf-8, newline) as f: f_csv csv.DictWriter(f, data[0].keys()) f_csv.writerows(data) print(end to write to file) def get_provinces(self): province_file input/province.csv if os.path.exists(province_file): return list(csv.DictReader(open(province_file, encodingutf-8))) provinces self.get_json(http://www.nmc.cn/f/rest/province) self.write_csv(province_file, provinces) return provinces def get_cities(self): city_file input/city.csv if os.path.exists(city_file): return list(csv.DictReader(open(city_file, encodingutf-8))) cities [] for province in self.get_provinces(): code province[url].split(/)[-1].split(.)[0] cities.extend(self.get_json(http://www.nmc.cn/f/rest/province/ code)) self.write_csv(city_file, cities) return cities def get_passed_weather(self): target_file input/passed_weather_ALL.csv all_weather [] count 0 for city in self.get_cities(): data self.get_json(http://www.nmc.cn/f/rest/passed/ city[code]) if not data: continue 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) all_weather.extend(data) time.sleep(1) if count % 50 0: self.write_csv(target_file, all_weather) all_weather [] if all_weather: self.write_csv(target_file, all_weather) if __name__ __main__: WeatherCrawler().get_passed_weather()逻辑说明get_provinces和get_cities都做了os.path.exists判断意思是第一次爬取后生成 CSV之后直接读取不用重新请求接口。get_passed_weather里每处理完 50 个城市就写一次盘避免全部数据积在内存里导致 OOM。参数说明time.sleep(1)是必要的爬 2412 个城市的天气数据一次性高频率请求容易被站点限流User-Agent必须伪装否则很多气象接口对默认 urllib 请求直接返回 403。3.3 数据落盘与文件清理最终input目录下会生成三个文件province.csv、city.csv、passed_weather_ALL.csv。passed_weather_ALL.csv每行代表某城市某整点的天气快照字段包括省份、城市名称、城市代码、时间点、气温、过去 1 小时降雨量。如果爬取中断直接删除passed_weather_ALL.csv重跑即可前两个文件不用动。有一个我在第一次跑就踩过的坑爬下来的 JSON 里time字段可能是空字符串或者None而写入 CSV 时用的是city_code等字段名如果在write_csv里直接用data[0].keys()后续 Spark 读表头时会发现某些记录字段对不上。所以我个人习惯在写盘之前先执行一行校验# 检查字段一致性防止部分城市数据缺失字段 all_weather [item for item in all_weather if time in item and rain1h in item]4. Spark SQL 分析计算 24 小时累计降水量与四时次平均气温4.1 读取 CSV 并完成类型转换数据进入 Spark 后的第一件事是类型转换。spark.read.csv默认把全部字段读成字符串所以在filter之前必须先 cast否则会出现气温 12.0 小于 1000这种永远为真的错误判断。from pyspark.sql import SparkSession from pyspark.sql.functions import col spark SparkSession.builder.master(local[2]).appName(weather-analyse).getOrCreate() # 读取爬取的 CSV 数据 df spark.read.csv(input/passed_weather_ALL.csv, headerTrue, encodingUTF-8) df.printSchema()注意headerTrue表示把 CSV 第一行当作列名encoding参数在中文数据集上要显式指定否则 Spark 默认按 UTF-8 解析如果 CSV 首行带 BOM 会导致第一列列名变成\ufeffprovince。我的经验是看到printSchema()里出现奇怪列名先检查 CSV 文件头部是否包含 BOM 字节。4.2 计算过去 24 小时累计降水量思路很直接按城市分组对rain1h求和。但中间要加一步异常值过滤。项目原文给出的界限是大于 1000 是异常数据因为气象站点每小时降雨量不可能超过 1000mm出现这种值多半是传感器故障或 JSON 字段对齐错误。from pyspark.sql.functions import col, sum, desc # 选择需要的字段并把 rain1h 转成 double 类型 df_rain df.select( province, city_name, city_code, col(rain1h).cast(double).alias(rain1h) ) # 过滤异常值 df_rain df_rain.filter(col(rain1h) 1000) # 按城市分组求和 df_rain_sum df_rain.groupBy(province, city_name, city_code) \ .agg(sum(rain1h).alias(rain24h)) \ .sort(desc(rain24h)) # 缓存结果后续多次取数不用重复计算 df_rain_sum.cache() # 结果展示 Top 10 df_rain_sum.show(10)groupBy(province, city_name, city_code)是联合分组Spark 会按这三个字段的组合做聚合保证同名城市在不同省份不会被混在一起。agg(sum(rain1h).alias(rain24h))是聚合阶段的核心给结果列起一个可读的名字。.sort(desc(rain24h))的作用是让结果按降雨量从高到低排列便于后续 Top-N 可视化。4.3 计算四时次日平均气温这一节是整套分析中最容易算错的模块。按照地面气象观测规范日平均气温不是简单地从 00 点到 23 点取平均而是取 02 时、08 时、14 时、20 时四个时次的数据求平均。我复现时特意在时间字段上做了hour()提取再从 4 个时次中按城市分组计算。from pyspark.sql.functions import date_format, hour, avg, count, format_number # 提取日期和小时字段 df_temperature df.select( province, city_name, city_code, col(temperature).cast(double).alias(temperature), date_format(col(time), yyyy-MM-dd).alias(date), hour(col(time)).alias(hour) ) # 只保留四时次数据 df_4point df_temperature.filter(col(hour).isin(2, 8, 14, 20)) # 按城市和日期分组同时做计数和平均值聚合 df_avg_temperature df_4point.groupBy(province, city_name, city_code, date) \ .agg( count(*).alias(cnt), avg(temperature).alias(avg_temperature) ) \ .filter(col(cnt) 4) \ .select( province, city_name, city_code, date, format_number(avg_temperature, 1).alias(avg_temperature) ) \ .sort(desc(avg_temperature)) df_avg_temperature.show(20)这里有个细节值得展开filter(col(cnt) 4)是防止只有两个时次就计算日平均温的情况。我在实际数据里发现有些城市在某个整点数据缺失如果直接 groupBy 后平均会把半天数据也算成日平均温结果偏离真实值。所以一定要先计数再过滤。date_format(col(time), yyyy-MM-dd)是把原始时间字符串截成只含日期因为 Spark 的time字段格式可能包含分秒比如2022-12-12 08:00:00我们需要的是按天分组所以只保留日期部分。format_number(avg_temperature, 1)是做四舍五入保留一位小数方便写进报告。4.4 缓存与结果落盘coalesce 到底是干什么的聚合结果如果要反复读取用于可视化建议在show之前调用cache()。Spark 的惰性求值意味着每次show、collect都会重新执行一遍完整的任务链数据量小的时候不明显但 57888 条数据加上多次 Top-N 展示时缓存能明显加快速度。# 结果持久化到本地文件 df_rain_sum.coalesce(1).write.csv(output/rain24h, headerTrue, modeoverwrite) df_avg_temperature.coalesce(1).write.json(output/avg_temperature, modeoverwrite)为什么用coalesce(1)而不是repartition(1)因为coalesce只会减少分区不会触发 Shuffle效率高得多。coalesce(1)把最终结果合并到一个分区再写出确保输出目录下只有一个 part 文件而不是分散的十几个文件这样拷贝到报告里很方便。写 CSV 时记得加headerTrue否则读输出的表头会缺失。5. 避坑记录这套环境最常见的五个翻车现场5.1 Java 版本不对Spark 直接闪退现象执行pyspark时终端只输出一行日志就退出或者 Jupyter 里创建SparkSession时报UnsupportedClassVersionError。原因Spark 2.4.0 编译时默认 target 是 Java 8如果系统里只有 JDK 11 或更高版本类加载时会因为版本不匹配出现找不到主类的情况。解决安装 OpenJDK 8 并显式设置JAVA_HOME。我建议在spark-env.sh里写死JAVA_HOME路径因为只改/etc/environment有可能在 Jupyter notebook 内被覆盖。5.2 Spark 2.4.0 与 Python 3.9 的兼容性问题现象pyspark启动时能正常打印版本号但一执行count()、show()就出现py4j.Py4JException: Constructor org.apache.spark...或socket connection refused。原因Spark 2.4.0 发布时 Python 3.9 还没有出来Py4J 的旧版本与 Python 3.9 的解释器在某些调用上不兼容。解决保持 Python 3.9 不变时在spark-env.sh和 shell 环境里都必须指定PYSPARK_PYTHON/usr/bin/python3.9然后确保 Jupyter 的内核同样使用这个解释器。最稳妥的做法是python3.9 -m pip install pyspark2.4.0这样 Jupyter 的 Python kernel 里 import 的 pyspark 和自己下载的 Python 包是同一个版本避免系统 Spark 目录里的 Python 包与 Jupyter kernel 冲突。如果换了仍报错我一般会退到pyspark2.4.8这是 2.4 系列最后一个小版本对 Python 3.9 支持最完整。5.3 读 CSV 时第一列列名多了\ufeff前缀现象df.printSchema()输出列名是\ufeffprovince导致df.select(province)报cannot resolve column name。原因爬虫写入 CSV 时用codecs.BOM_UTF8写入了文件头 BOM 字节Spark 的 CSV parser 在encodingUTF-8下默认不剥离 BOM。解决爬虫代码里不要调用f.write(codecs.BOM_UTF8)Python 3 的open(file, w, encodingutf-8)本身不带 BOM。如果文件已经生成用下面这行修正列名df df.withColumnRenamed(\ufeffprovince, province)5.4 爬虫报urllib3.Request不存在现象执行爬虫时提示AttributeError: module urllib3 has no attribute Request。原因原项目代码误用了urllib3.Request这个模块本身就没有 Request 类。在 Python 3 里标准库的正确模块是urllib.request不是urllib3。解决改成urllib.request.Request(url)并加req.add_header(User-Agent, ...)。顺便说一句time.sleep(1)不能省否则爬到几百个城市时很容易被服务器限流返回空 JSON导致if data:直接跳过大量城市。5.5coalesce(1)写出的不是单个文件而是目录现象查看output/rain24h时发现里面有一堆part-00000-xxx文件报告里没法直接引用一个干净的文件。原因Spark 的write.csv语义是写出到一个目录目录内包含多个分区文件。coalesce(1)只是把分区数缩成 1不改变目录part 文件的落盘结构。解决输出后用一个命令把 part 文件重命名复制出来cat output/rain24h/part-*.csv output/rain24h_result.csv不过我一般建议保留 part 文件结构因为 Spark SQL 语义下目录里同时还有一个_SUCCESS文件这是任务成功的标记报告里截图展示这个目录反而更能证明分析跑通了。6. 装完怎么验证从 Jupyter 跑完到 Top-N 排名和可视化图表最后落地到验证环节我的习惯是拿到聚合结果后不急着画图先做三件事确认行数、确认 schema、验证 Top-N 值是否符合直觉。先跑一遍基本校验代码# 1. 读回写出的结果确认数据能正常回读 df_result spark.read.csv(output/rain24h, headerTrue) print(总行数:, df_result.count()) df_result.printSchema() # 2. 用 collect 把结果拉回 driver转成列表供 matplotlib 用 rain_top df_rain_sum.limit(20).collect() print(rain_top[:5])这里顺序有讲究先用show(10)看是不是全 0 或全 NULL。气象数据里大量城市整点降雨量本来就是 0如果 Top-N 全是 0说明字段类型转换出了问题比如rain1h被当作字符串求和Spark 对非数字字符串求和会静默返回 0。确认无误后再做可视化。我通常会画出降雨量 Top-20 的直方图用中文城市名做 x 轴标签这一步能直观验证数据抓取和聚合是否形成闭环import matplotlib.pyplot as plt cities [r[city_name] for r in rain_top] values [float(r[rain24h]) for r in rain_top] plt.figure(figsize(12, 6)) plt.bar(cities, values) plt.xticks(rotation45) plt.title(past 24h rainfall top 20) plt.ylabel(rainfall (mm)) plt.tight_layout() plt.savefig(output/rain_top20.png, dpi120) plt.show()图表能画出来、文件名里包含预期城市、柱状图趋势不是一条直线这套基于 Spark 的气象数据处理与分析项目就真正跑通了。从那以后我每接手一套 Spark 环境都会强制自己先跑一遍spark-submit --version、再在 Jupyter 里执行一个最小的spark.range(10).count()确认版本、JDK、Python 三者没有隐性问题后才开始写分析代码。希望这篇安装笔记帮到你至少让你少走几个我走过的弯路。本文还有配套的精品资源点击获取