基于Spark的气象数据处理与分析：从爬虫到可视化完整实践

简介面向大数据期末课设的Spark气象数据处理与分析完整报告适合需要完成Spark实战项目、掌握气象数据分析全流程的大数据专业学生参考。项目基于中央气象台2412个城市、57888条真实天气数据完整呈现从JSON异步数据爬取、CSV存储、Spark DataFrame清洗聚合到matplotlib可视化的实现过程并给出省份、城市、天气三级接口的解析思路与爬虫源代码。报告按课程设计标准撰写章节涵盖项目背景、实验环境、数据来源、数据获取、数据分析与可视化其中重点演示了分组聚合、缺失异常值处理、窗口函数等Spark常用操作可作为课程报告或答辩展示的实用模板。内含1个docx文档744KB已有13405人学习适合大数据入门者对照演练。1. 基于Spark的气象数据处理与分析一份能直接复用的大数据期末课设每年期末总有同学卡在同一个问题上Spark课设到底做什么才能既体现技术含量又能在两周内跑完这份基于Spark的气象数据处理与分析课设项目走的就是一条非常务实的路线——爬取全国2412个城市、57888条整点天气数据用Spark SQL完成24小时累计雨量和日均气温的聚合计算再用matplotlib把结果画出来。整套流程覆盖了数据获取、清洗、分析、可视化四个环节技术栈就是Spark DataFrame API加Python没有绕圈子也没有堆砌高深算法。适合大数据专业期末课设、Spark入门实战以及想了解“用Spark做气象数据处理”完整链路的人。下文我按实际落地顺序把环境、爬虫、预处理、分析和踩坑点逐一拆开讲。2. 环境与数据获取从三级URL到57888条天气数据的爬取链路2.1 实验环境选型与组件安装这个课设的实验环境是一个典型的单机Spark开发环境不需要搭建集群。我复现时使用的组件版本如下表所示组件版本说明LinuxUbuntu 16.04课设报告中指定的操作系统Python3.9用于爬虫、Spark任务和可视化Spark2.4.0计算引擎跑在local模式下Jupyter Notebook最新版交互式调试DataFrame和绘图有一点值得注意Spark 2.4.0与Python 3.9的组合在实际使用中会有兼容性警告网上很多资料说Spark 2.4.0不支持Python 3.8以上的版本实际跑起来Spark SQL基础功能问题不大但如果你用到pandas UDF或新的API建议直接换Spark 3.x。课设场景下沿用2.4.0没有问题只要别在它上面做太新奇的特性就行。装完Spark和Jupyter之后还需要给Python补上可视化的依赖sudo apt-get install python3-matplotlib sudo apt-get install python3-tkpython3-matplotlib是绘图库本体python3-tk是matplotlib在本地窗口显示图表时需要的Tkinter后端。如果你用的是无显示器的服务器还应在代码开头加上matplotlib.use(Agg)否则绘图时找不到显示设备会直接报错。Ubuntu 16.04的apt源里自带这两个包优先用它装比pip装更省事。2.2 三级URL接口设计先拿省份再拿城市最后拿天气气象数据的来源是某公开气象服务站点。通过浏览器开发者工具观察网络请求能发现数据不是渲染在HTML页面里的而是通过异步请求直接返回JSON字符串。核心的请求路径可以归纳为下面三条层级请求路径返回内容省份列表/f/rest/province省份名称与省份三位编码城市列表/f/rest/province/{省份三位编码}该省所有城市的名称与城市编号整点天气/f/rest/passed/{城市编号}该城市最近24小时整点天气数组三级接口是典型的嵌套依赖关系省份编码必须从第一级接口的返回里取城市编号必须从第二级接口的返回里取第三级接口才能拿到真正的气象数据。比如某个省在省份接口里的编码是“ABC”请求城市列表时要带上这个编码返回的城市数据里又有类似“58847”的城市编号把它拼到第三级URL末尾才能拿到该城市24小时整点数据。爬虫就是要顺着这条链子把全国城市依次走一遍。2.3 爬虫代码实现与CSV分段落盘爬虫部分原课设报告用的是Python 2时期的urllib3.Request写法但urllib3库里并没有Request类直接跑会报AttributeError。我改成Python 3标准库的urllib.request逻辑保持一致import urllib.request import json import csv import codecs import os import time class WeatherCrawler: def __init__(self, base_url): self.base_url base_url def fetch_json(self, path): url self.base_url path req urllib.request.Request(url, headers{User-Agent: Mozilla/5.0}) with urllib.request.urlopen(req, timeout10) as resp: return json.loads(resp.read().decode(utf-8)) def write_csv(self, file_name, rows, write_headerFalse): if not rows: return with open(file_name, ab) as f: f.write(codecs.BOM_UTF8) writer csv.DictWriter(f, fieldnamesrows[0].keys()) if write_header: writer.writeheader() writer.writerows(rows)fetch_json里设置了User-Agent请求头避免站点拒绝默认的Python urllib请求。timeout10是超时控制防止某个城市接口卡住导致整个爬虫挂起。write_csv里写入codecs.BOM_UTF8是为了让Excel直接打开CSV时中文不乱码write_header参数控制是否写表头因为分段落盘时只有第一个文件块需要写表头。下面是三级数据获取的完整调度逻辑def get_provinces(self): if os.path.exists(input/province.csv): return self.read_csv(input/province.csv) provinces self.fetch_json(/f/rest/province) self.write_csv(input/province.csv, provinces, write_headerTrue) return provinces def get_cities(self): if os.path.exists(input/city.csv): return self.read_csv(input/city.csv) cities [] for province in self.get_provinces(): code province[url].split(/)[-1].split(.)[0] cities.extend(self.fetch_json(/f/rest/province/ code)) time.sleep(1) self.write_csv(input/city.csv, cities, write_headerTrue) return cities def get_passed_weather(self): cities self.get_cities() passed_weather [] count 0 for city in cities: data self.fetch_json(/f/rest/passed/ city[code]) if data: count 1 for item in data: item[city_code] city[code] item[province] city[province] item[city_name] city[city] item[city_index] str(count) passed_weather.extend(data) time.sleep(1) if count % 50 0 and count 0: self.write_csv(input/passed_weather_ALL.csv, passed_weather, write_header(count 50)) passed_weather [] if passed_weather: self.write_csv(input/passed_weather_ALL.csv, passed_weather, write_header(count 50))这段代码有两个关键设计。第一个是幂等province.csv和city.csv已存在时直接读取全量天气数据已经落盘后重新运行也不会重复爬取想刷新数据就手动删掉这三个文件。第二个是每爬50个城市就落盘一次——全国2412个城市每条数据有多个整点记录如果全部堆在内存里等到最后再一次性写文件爬半小时内存早就爆了。分段写入之后即使中间某个城市请求超时导致中断已爬的部分也不会丢。3. Spark SQL数据预处理Schema推断、类型转换与异常过滤3.1 SparkSession初始化与CSV读取数据爬下来之后下一步就是用Spark读进来。这里用的是Spark SQL中的DataFrame API而不是RDD——DataFrame有类型信息SQL风格表达清晰而且Spark默认的Catalyst优化器会自动优化执行计划对课设这种量级的数据完全够用。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WeatherAnalysis) \ .master(local[4]) \ .getOrCreate() df spark.read.csv( input/passed_weather_ALL.csv, headerTrue, inferSchemaTrue )master(local[4])表示本地模式用4个线程并行执行对六万条数据来说绰绰有余。headerTrue把CSV第一行作为列名inferSchemaTrue让Spark扫描数据后自动推断每一列的类型。这样temperature和rain1h会被识别成double后续做数值计算不用手动转换。但inferSchema有一个坑如果CSV某列的前面一部分行是缺失值或非数字字符串Spark可能把整列推断成string后面用比较运算符直接报错。稳妥的做法是在读取后打印Schema检查一遍df.printSchema() df.show(5, truncateFalse)printSchema()输出的每一行格式是“列名 类型”一眼就能看出有没有推断错误。show(5, truncateFalse)则是查看前5条原始数据truncateFalse确保长字段不会被截断方便直接观察time字段的格式。3.2 字段选择与数值类型转换原课设报告里有一个很容易被忽略的细节CSV经过Python爬虫写出后即使Spark用inferSchema读入某些列的类型仍是字符串。更稳妥的做法是在分析前显式做一次类型转换把注意力集中在真正需要的列上from pyspark.sql.functions import col df_rain df.select( province, city_name, city_code, rain1h ).filter( col(rain1h).cast(double) 1000 )select只取四个字段把数据宽度压缩到最小col(rain1h).cast(double)强制把降雨量转成数值型这样即使某一行是空字符串也不会让整个作业报错而是转为null。过滤条件 1000是排除异常值——正常城市24小时累计降雨量不可能超过1000毫米一旦出现这种记录基本是站点数据采集故障或爬取时拿到的脏数据。3.3 时间字段解析与温度数据准备温度分析需要把time字段拆成“日期”和“小时”两部分这就要用到pyspark.sql.functions里的两个函数from pyspark.sql.functions import date_format, hour df_temp df.select( province, city_name, city_code, temperature, time ).withColumn( date, date_format(col(time), yyyy-MM-dd) ).withColumn( hour, hour(col(time)) )date_format(col(time), yyyy-MM-dd)从时间戳里截取日期部分hour(col(time))提取小时数值。这里的前提是CSV里time字段能被Spark识别为时间类型。如果Spark读进来之后time是字符串这两个函数会返回null。遇到这种情况先看原始数据的time字段格式再决定是加一个to_timestamp(col(time), yyyy-MM-dd HH:mm:ss)转换还是直接用字符串截取函数处理。整个预处理阶段的目标就是得到两张干净的中间表一张是带rain1h数值的df_rain一张是带date和hour字段的df_temp。两张都是懒加载的DataFrame真正触发计算的是后续的groupBy聚合操作。4. 两个核心分析任务24小时累计雨量与四时次日均气温4.1 累计降雨量分析groupBy、agg与全局排序降雨量分析的思路很直接按省份和城市分组对rain1h字段求和得到每个城市过去24小时的累计雨量然后降序排列取Top-N。Spark代码是这样的df_rain_sum df_rain.groupBy( province, city_name, city_code ).agg( {rain1h: sum} ).withColumnRenamed( sum(rain1h), rain24h ).sort( col(rain24h).desc() )agg({rain1h: sum})是传入字典的写法键是列名值是聚合函数结果列名自动生成为sum(rain1h)所以紧接着用withColumnRenamed重命名成rain24h后面引用起来更直观。.sort(col(rain24h).desc())是分布式环境下的全局排序Spark会做一次shuffle把所有分区数据按雨量倒排对2400多个城市的数据量来说代价很小。排序之后head(10)可以直接取到雨量最大的十个城市这个动作会触发Spark真正执行前面的所有转换步骤top_rain df_rain_sum.head(10) for row in top_rain: print(row[city_name], row[rain24h])注意head(10)返回的是一个Row对象列表每个Row就像是Python里的字典用row[列名]取值。这一步执行时Spark会把懒加载的整个血缘关系链从CSV读取一路算到排序所以如果前面某一步写错了报错是在这里炸出来而不是在groupBy那里。4.2 日均气温分析四时次筛选与聚合计算日均气温这一段是项目里最有气象业务味道的地方。根据《地面气象服务观测规范》日平均气温取的是02时、08时、14时、20时四个时次气温的平均值而不是24小时整点气温的算术平均。这是气象行业的规范直接决定了代码里的过滤条件from pyspark.sql.functions import count, sum as spark_sum, format_number df_4point df_temp.filter( col(hour).isin([2, 8, 14, 20]) ) df_avg df_4point.groupBy( province, city_name, city_code, date ).agg( count(temperature).alias(cnt), spark_sum(temperature).alias(temp_sum) ).filter( col(cnt) 4 ).withColumn( avg_temperature, format_number(col(temp_sum) / 4.0, 1) ).select( province, city_name, city_code, date, avg_temperature )整个流水线的关键是先计数再求均值count(temperature)看看每个城市在目标日期里是否四个时次的数据都齐了只有cnt 4的记录才参与平均计算。如果某个城市某天只有3个时次的数据直接算平均气温会和真实值有偏差过滤掉才是严谨的做法。format_number(..., 1)把结果格式化成带一位小数的字符串方便直接用于可视化展示。这个查重逻辑在课设报告里已经说得很细但我复现时发现一个数据层面的问题爬虫是在白天某个时刻运行的当天20时还没到那么当天20时的记录根本不存在。原报告也提示了“建议在当天20时30分之后再爬取数据”这说明分析结果的质量高度依赖爬取时机。4.3 cache与coalesce性能优化和输出控制两个分析任务在最后都调了cache()和coalesce(1)这一步值得展开讲df_rain_sum.cache().coalesce(1).write.csv(output/rain_sum, headerTrue) df_avg.coalesce(1).write.json(output/avg_temp, headerFalse)cache()把前面经过groupBy和sort的中间结果物化到内存里这样后面如果既要做head(10)又要write.csvSpark不用从头再算两遍。coalesce(1)是把分区数量压缩到1个这样CSV输出只有一个文件而不是默认的几十个part-00000碎片后续直接读这个单文件就行。但coalesce(1)不是免费的。它本质上是一次结果重分区把所有分区的数据拉到同一个分区再写出数据量大时会产生单点瓶颈。雨量Top-N分析加coalesce(1)没问题但如果以后换成每天几十GB的气象数据正确做法是保留分区数并写成parquet格式不要强行合并成单文件。这个边界要拎清楚。5. 常见问题与排查爬虫、Spark和可视化里的五个坑5.1 Python 3下运行报错sys.setdefaultencoding不存在现象把课设原爬虫代码直接丢到Python 3环境运行启动就报AttributeError: module sys has no attribute setdefaultencoding。原因sys.setdefaultencoding(utf8)是Python 2时代处理默认编码的命令Python 3的默认编码就是UTF-8这个函数已经被移除。原课设写的是reload(sys)和sys.setdefaultencoding(utf8)这是典型的Python 2代码。解决直接删掉这两行。如果担心中文编码有问题用Python 3在打开文件时指定encodingutf-8就够了不需要再动全局默认编码。这个坑几乎是每届课设必现看到这个报错就知道代码是从旧版本迁移过来的。5.2 爬虫结果CSV用Excel打开中文乱码现象passed_weather_ALL.csv在终端和代码里读着都正常用Excel双击打开时中文全部变成乱码省份名和城市名完全不可读。原因CSV文件没有BOM头。Python默认用UTF-8写文本文件不掺BOM而Excel的CSV解析器在识别非UTF-8编码时倾向用ANSI解码中文结果就是乱成一团。原报告里在write_csv中写入codecs.BOM_UTF8就是为了解决这个问题。解决写入CSV前先写一次codecs.BOM_UTF8这是给Excel看的编码标记。注意每次打开文件追加写入时都要判断文件是否已存在如果已存在就不要重复加BOM否则表头里会多出隐藏的非法字符。5.3 Spark读CSV后数值列变成字符串过滤直接报错现象df.filter(col(rain1h) 1000)执行时抛出分析异常提示无法用字符串和数字比较打印Schema发现rain1h列的类型是string而非double。原因inferSchemaTrue不是万能的。如果CSV的前若干行里rain1h字段是空值或占位符比如--Spark在推断类型时会认为整列是字符串。尤其是爬虫写CSV时把数字转成了字符串空字段就是空字符串更容易触发误判。解决最直接的办法是用col(rain1h).cast(double)做显式转换再配合filter过滤。另一种做法是给read.csv显式传schema参数例如StructType里指定rain1h为DoubleType()彻底跳过推断机制。从工程习惯来说显式指定Schema比inferSchema更可靠但课设阶段写清楚cast就够了。5.4 日均气温结果城市数大幅缩水只有几百个城市现象计算完日均气温后输出结果里的城市数远少于数据源里的2412个城市少了一半还多。检查原始数据发现每个城市确实有整点气温数据。原因filter(col(cnt) 4)把只拥有部分时次记录的城市全部剔除了。爬虫运行时间是白天当天20时的数据不存在有些城市02时或08时的记录缺失凑不齐四时次的都被过滤掉了。解决四种选择。一是按原报告建议在当天20时30分之后再爬取保证当天四个时次都齐二是把cnt 4放宽到cnt 3并在报告里说明“基于三时次近似计算日均气温”适合只想跑通流程的情况三是查看缺失时次的分布规律找出是哪几个时次最容易缺四是修改爬虫逻辑补充历史数据接口而不是只爬最近24小时。课设报告采用的是第一种输出的质量是最高的。5.5 结果目录里出现一堆part文件没有单个CSV文件现象执行df_rain_sum.write.csv(output/rain_sum)后打开目录看到的是part-00000-xxx.csv、part-00001-xxx.csv十几个文件而不是想象中的rain_sum.csv。原因Spark分布式写入时每个分区各自写一个文件分区数就是输出文件数。默认情况下shuffle之后分区数较多自然会产生多个part文件。这不算错误但对后续人工查看数据很不方便。解决写之前加coalesce(1)强制把分区数减到1输出就只有一个part文件了。更省事的办法是不加coalesce(1)读取时直接spark.read.csv(output/rain_sum)读整个目录Spark会自动合并多个文件的内容。两种方式我都用过课设评分时如果要求提交“干净的CSV文件”就选coalesce(1)。6. 结果验证与可视化输出用matplotlib把分析结果变成图表数据分析算完不算结束还要验证结果正确然后再画图。验证有一个很土但很有效的办法——抽样对账。我从df_rain_sum.head(10)里随机挑了三个城市找到它们在passed_weather_ALL.csv里的所有rain1h记录手工在Excel里求和和Spark算出的rain24h对比。三次都对得上说明聚合逻辑没问题。这一步强烈建议保留它能把“代码跑通”和“结果正确”区分开答辩时也更有底气。可视化部分代码比较简短用matplotlib画横向条形图展示雨量Top-10城市import matplotlib.pyplot as plt top df_rain_sum.head(10) names [row[city_name] for row in top] values [float(row[rain24h]) for row in top] plt.rcParams[font.sans-serif] [SimHei, WenQuanYi Zen Hei] plt.rcParams[axes.unicode_minus] False plt.figure(figsize(10, 6)) plt.barh(names[::-1], values[::-1]) plt.xlabel(24小时累计降雨量mm) plt.title(过去24小时累计降雨量 Top10 城市) plt.tight_layout() plt.savefig(output/rain_top10.png, dpi150)plt.rcParams两行是解决中文字符显示成方块的设置SimHei是Windows下的黑体Linux下没有时就回退到WenQuanYi Zen Hei。names[::-1]和values[::-1]是把列表倒序让雨量最大的城市显示在条形图最上方符合阅读习惯。savefig的dpi150保证图片在答辩投屏时不模糊。气温分析的结果可以用类似思路画一张散点图或直方图查看全国城市日均气温的分布形态。到这一步从爬虫、清洗、聚合到出图整条链路就闭环了。想起那次复现后半段我发现某个城市计算出日均气温为97度一查是20时的气温字段爬到了异常值温度列混入了噪点数据。从那以后我每次做数据类课设都会强制走一遍“抽样对账”再进入可视化这个习惯帮我挡掉了至少三次答辩翻车。希望这份基于Spark的气象数据处理与分析课设梳理能帮你也少踩几个同样的坑。本文还有配套的精品资源点击获取