基于Python Flask构建天气数据采集与可视化分析系统

简介数据采集与可视化是数据分析领域的基础环节其核心原理是通过自动化程序从网络获取数据并利用图表、地图等形式直观呈现数据规律。这项技术对于挖掘数据价值、辅助决策至关重要广泛应用于市场分析、环境监测、业务监控等场景。本文聚焦于一个具体的工程实践案例探讨如何运用Python Flask框架、SQLAlchemy ORM以及ECharts、Leaflet等库构建一个集数据爬取、存储、API服务和多维度可视化于一体的完整系统。文中详细阐述了应对网络请求、数据解析、数据库设计及前端渲染等环节的挑战特别是如何设计健壮的爬虫程序和处理海量数据为开发类似的数据驱动型应用提供了可复用的解决方案。1. 项目概述与核心价值最近在做一个挺有意思的玩意儿一个针对山东省的天气数据爬取和可视化分析系统。起因很简单我之前想查查老家山东某个小县城过去几年的天气趋势发现要么数据太零散要么就是得付费免费的接口要么限制多要么数据维度不全。作为一个喜欢折腾数据的Python开发者这事儿就琢磨上了能不能自己动手把山东省各地市的实时天气、历史气象数据都抓下来存到自己的数据库里然后想怎么分析就怎么分析想怎么展示就怎么展示于是就有了这个基于Python Flask框架的项目。这个系统说白了就是一个集数据采集、存储、分析和可视化于一身的“一站式”天气数据工坊。它的核心价值在于把公开但分散的气象信息通过自动化的爬虫程序规整起来再利用Web技术直观地呈现出来。你不仅可以随时查看全省的实时天气状况还能回溯历史分析温度、湿度、风速、降水这些关键气象要素的变化规律无论是看长期气候趋势还是对比不同城市间的天气差异都变得非常方便。它特别适合对气象数据感兴趣的个人开发者、数据分析初学者或者需要本地化天气数据做研究的学生朋友。你不用再在各个网站间反复横跳所有数据和分析视图都在你自己的服务器上。2. 系统整体架构与设计思路2.1 技术栈选型与考量整个系统的构建我选择了一个非常经典且轻量的Python Web技术组合Flask作为后端框架SQLite后期可平滑迁移至MySQL/PostgreSQL作为数据存储前端则主要依靠ECharts和Leaflet这两个强大的JavaScript库进行可视化渲染。选择Flask而不是Django主要是出于灵活性和项目规模的考虑。这个天气系统核心是数据管道和可视化并不需要Django那种“全家桶”式的高度集成ORM、Admin后台。Flask的微框架特性让我可以按需引入组件比如用Flask-SQLAlchemy处理数据库用Flask-CORS解决可能的跨域问题用APScheduler做定时爬取任务。这种“搭积木”的方式让项目结构非常清晰也更容易控制。数据爬取部分核心是requests库负责网络请求BeautifulSoup4和lxml负责解析HTML页面。这里有个关键点我并没有完全依赖于某一个单一的免费天气API因为很多API有调用频率、历史数据深度或数据完整性的限制。我的策略是“多源互补”结合了中国天气网、心知天气免费额度等几个数据源通过校验和互补来确保数据的可靠性。比如实时数据可能以A源为主历史数据则从B源补全。可视化层面ECharts的强大足以应对折线图、柱状图、饼图等常规统计图表展示温度变化、降水量对比等趋势。而地图相关的可视化特别是热力图Leaflet是更专业的选择它能轻松地将数据与地理坐标结合展示全省的温度分布、降水强度等空间信息。2.2 系统模块拆解整个系统可以清晰地划分为四个核心模块它们通过Flask应用进行串联数据采集模块这是系统的“触手”。包含实时数据爬虫子模块和历史数据爬虫子模块。实时爬虫会以较高的频率如每30分钟运行抓取山东省各主要城市的最新天气状况历史爬虫则通常在凌晨低峰期运行批量补全或更新过去的气象数据。所有爬虫都被封装成独立的函数或类并通过APScheduler进行定时调度。数据存储与管理模块这是系统的“记忆库”。使用SQLAlchemy定义数据模型主要包含两张核心表real_time_weather存储实时数据如当前温度、湿度、风速、天气现象、更新时间和historical_weather存储历史数据除基本要素外还包括最高/最低温、降水量、风向等。设计表结构时充分考虑了查询效率比如对城市名称、日期等字段建立了索引。后端API服务模块这是系统的“中枢神经”。由Flask提供一系列RESTful API接口。例如/api/real_time/city_name获取指定城市实时天气/api/history?city济南start_date2023-01-01end_date2023-12-31查询历史数据/api/city_list获取所有支持的城市列表。这些接口为前端可视化提供纯净的数据源。前端可视化展示模块这是系统的“脸面”。一个简单的Flask模板使用Jinja2渲染出主页面页面内通过Ajax调用后端API获取JSON数据然后驱动ECharts和Leaflet生成图表和地图。主要视图包括全省实时天气地图带标记点、单城市多要素历史趋势折线图、多城市数据对比柱状图、气象要素分布热力图等。注意合规性与道德爬虫在开发爬虫时务必遵守目标网站的robots.txt协议合理设置请求间隔如添加time.sleep(random.uniform(1, 3))避免对目标服务器造成压力。尽量使用公开API如果有解析网页是次选方案。本项目主要用于个人学习与技术实践。3. 核心实现细节与实操要点3.1 数据爬虫的稳健性设计爬虫是数据质量的源头必须足够健壮。我以爬取中国天气网济南市历史页面为例讲解几个关键设计。首先请求头Headers的模拟至关重要需要让请求看起来像一个普通的浏览器访问。import requests from bs4 import BeautifulSoup import time import random headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept-Language: zh-CN,zh;q0.9, } def fetch_jinan_history(year, month): 抓取济南某年某月的历史天气数据 url fhttp://www.weather.com.cn/weather40d/101120101.shtml # 示例URL实际需查找 try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() # 检查HTTP状态码是否为200 resp.encoding utf-8 # 统一编码 return resp.text except requests.exceptions.RequestException as e: print(f请求失败: {e}) log_error_to_db(url, str(e)) # 将错误记录到数据库便于排查 return None其次页面解析要考虑到网站结构可能发生变化。使用BeautifulSoup配合lxml解析器需要安装lxml库效率更高。解析时不能写死标签路径应尽量寻找具有唯一性的id或class属性。def parse_history_html(html): 解析历史天气HTML if not html: return [] soup BeautifulSoup(html, lxml) data_list [] # 假设历史数据在一个id为history-data的表格中 history_table soup.find(table, {id: history-data}) if not history_table: # 如果找不到尝试其他选择器这里体现了容错性 history_table soup.find(div, class_history-table) if history_table: rows history_table.find_all(tr)[1:] # 跳过表头 for row in rows: cols row.find_all(td) if len(cols) 5: # 确保有足够列 date cols[0].text.strip() high_temp cols[1].text.strip().replace(℃, ) low_temp cols[2].text.strip().replace(℃, ) weather cols[3].text.strip() wind cols[4].text.strip() # 数据清洗和转换 try: data_list.append({ date: date, high_temp: float(high_temp) if high_temp else None, low_temp: float(low_temp) if low_temp else None, weather: weather, wind: wind }) except ValueError as e: print(f数据转换错误 {date}: {e}) continue return data_list实操心得数据清洗与异常处理原始数据往往很“脏”包含“℃”、“mm”、“级”等单位符号以及“-”、“/”等缺失值标识。必须在入库前进行清洗。我通常会写一个统一的清洗函数处理各种边缘情况。此外网络请求失败、页面结构变更、数据格式异常都是常态必须用try...except包裹核心逻辑并记录详细的错误日志方便后续补爬和系统维护。3.2 数据库模型设计与优化使用Flask-SQLAlchemy定义ORM模型清晰且易于操作。from datetime import datetime from flask_sqlalchemy import SQLAlchemy db SQLAlchemy() class City(db.Model): 城市基础信息表 id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(50), uniqueTrue, nullableFalse, indexTrue) # 城市名加索引 code db.Column(db.String(20), uniqueTrue) # 城市代码用于爬虫 latitude db.Column(db.Float) # 纬度用于地图 longitude db.Column(db.Float) # 经度用于地图 class RealTimeWeather(db.Model): 实时天气数据表 id db.Column(db.Integer, primary_keyTrue) city_id db.Column(db.Integer, db.ForeignKey(city.id), nullableFalse, indexTrue) temperature db.Column(db.Float) # 温度 humidity db.Column(db.Integer) # 湿度百分比 wind_speed db.Column(db.Float) # 风速 wind_direction db.Column(db.String(20)) # 风向 weather db.Column(db.String(50)) # 天气现象 pressure db.Column(db.Float) # 气压 visibility db.Column(db.Float) # 能见度 update_time db.Column(db.DateTime, defaultdatetime.utcnow, indexTrue) # 数据更新时间 city db.relationship(City, backrefdb.backref(realtime_records, lazydynamic)) class HistoricalWeather(db.Model): 历史天气数据表 id db.Column(db.Integer, primary_keyTrue) city_id db.Column(db.Integer, db.ForeignKey(city.id), nullableFalse, indexTrue) date db.Column(db.Date, nullableFalse, indexTrue) # 日期 high_temp db.Column(db.Float) # 最高温 low_temp db.Column(db.Float) # 最低温 avg_temp db.Column(db.Float) # 平均温可计算得出 precipitation db.Column(db.Float) # 降水量 avg_humidity db.Column(db.Integer) # 平均湿度 avg_wind_speed db.Column(db.Float) # 平均风速 weather_day db.Column(db.String(50)) # 白天天气 weather_night db.Column(db.String(50)) # 夜间天气 city db.relationship(City, backrefdb.backref(history_records, lazydynamic)) # 建立联合唯一索引防止同一城市同一日期的数据重复 __table_args__ (db.UniqueConstraint(city_id, date, nameuix_city_date),)设计考量数据去重在HistoricalWeather表上建立了(city_id, date)的联合唯一约束这是防止爬虫重复插入数据的核心机制。在插入数据前可以使用db.session.merge()或先查询后判断的方式避免冲突。索引优化对高频查询字段如city_id、date、update_time建立索引能极大提升查询速度尤其是在数据量积累到数十万条后效果显著。关系关联通过外键和relationship关联城市表这样在查询时可以直接通过weather.city.name获取城市名无需手动联表简化了操作。3.3 Flask API 设计与数据交互Flask后端提供清晰的数据接口。这里以历史数据查询API为例。from flask import Flask, request, jsonify from sqlalchemy import and_ from datetime import datetime, timedelta app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] sqlite:///weather.db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db.init_app(app) app.route(/api/history, methods[GET]) def get_history_weather(): 获取历史天气数据API city_name request.args.get(city, ) start_date_str request.args.get(start_date, ) end_date_str request.args.get(end_date, ) # 参数校验 if not city_name or not start_date_str or not end_date_str: return jsonify({error: Missing required parameters: city, start_date, end_date}), 400 try: start_date datetime.strptime(start_date_str, %Y-%m-%d).date() end_date datetime.strptime(end_date_str, %Y-%m-%d).date() except ValueError: return jsonify({error: Invalid date format. Use YYYY-MM-DD.}), 400 if start_date end_date: return jsonify({error: start_date must be earlier than end_date.}), 400 # 查询数据库 try: city City.query.filter_by(namecity_name).first() if not city: return jsonify({error: fCity {city_name} not found.}), 404 records HistoricalWeather.query.filter( and_( HistoricalWeather.city_id city.id, HistoricalWeather.date start_date, HistoricalWeather.date end_date ) ).order_by(HistoricalWeather.date.asc()).all() # 序列化数据 data [{ date: record.date.isoformat(), high_temp: record.high_temp, low_temp: record.low_temp, avg_temp: record.avg_temp, precipitation: record.precipitation, avg_humidity: record.avg_humidity, weather: f{record.weather_day}/{record.weather_night} } for record in records] return jsonify({city: city_name, data: data}) except Exception as e: app.logger.error(fDatabase query error: {e}) return jsonify({error: Internal server error}), 500这个API设计了完整的参数校验、错误处理和日志记录。返回的JSON结构清晰前端ECharts可以直接使用。对于可能返回大量数据的查询可以考虑加入分页参数page,per_page来优化性能。4. 多维度数据可视化实现4.1 全省实时天气地图Leaflet 自定义图标前端页面引入Leaflet库和山东省的GeoJSON边界数据。通过调用/api/real_time/all接口获取所有城市的实时数据然后在地图对应坐标上添加标记点。!-- 在Flask模板中 -- div idmap styleheight: 600px;/div script var map L.map(map).setView([36.5, 118.0], 7); // 以山东为中心 L.tileLayer(https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png, { attribution: © OpenStreetMap contributors }).addTo(map); // 加载山东边界 fetch(/static/geojson/shandong.json) .then(response response.json()) .then(data L.geoJSON(data).addTo(map)); // 获取实时数据并添加标记 fetch(/api/real_time/all) .then(response response.json()) .then(cities { cities.forEach(city { // 根据温度定义图标颜色 var temp city.temperature; var color temp 30 ? #ff0000 : temp 20 ? #ffa500 : temp 10 ? #ffff00 : #00ffff; var icon L.divIcon({ html: div stylebackground:${color}; border-radius:50%; width:20px; height:20px; text-align:center; line-height:20px; color:white; font-weight:bold;${Math.round(temp)}/div, iconSize: [20, 20] }); L.marker([city.lat, city.lon], {icon: icon}) .bindPopup(b${city.name}/bbr温度: ${city.temperature}℃br湿度: ${city.humidity}%br天气: ${city.weather}) .addTo(map); }); }); /script4.2 历史趋势折线图与多城市对比ECharts这是最常用的分析视图。使用ECharts的折线图可以清晰展示单个城市温度、湿度的年度/月度变化。而柱状图则适合对比不同城市在相同时间段内的平均降水量或高温天数。// 假设已获取到济南市2023年夏季的数据 historyDataJinan // 和青岛市同期数据 historyDataQingdao var chartDom document.getElementById(chart); var myChart echarts.init(chartDom); var option { title: { text: 2023年夏季济南 vs 青岛 最高温度对比 }, tooltip: { trigger: axis }, legend: { data: [济南最高温, 青岛最高温] }, xAxis: { type: category, data: historyDataJinan.map(item item.date) // 日期数组 }, yAxis: { type: value, name: 温度(℃) }, series: [ { name: 济南最高温, type: line, data: historyDataJinan.map(item item.high_temp), smooth: true }, { name: 青岛最高温, type: line, data: historyDataQingdao.map(item item.high_temp), smooth: true } ] }; myChart.setOption(option);4.3 气象要素空间分布热力图Leaflet.heat要展示如“全省夏季平均气温分布”这类空间连续数据热力图是最佳选择。我们可以使用Leaflet的插件leaflet.heat。首先需要准备数据计算每个城市或更细粒度网格点在目标时间段内的平均气温形成一个[lat, lng, intensity]的数组。// 假设 heatData 是从后端API获取的数组格式如 [[36.65, 117.12, 26.5], [36.09, 120.38, 24.1], ...] fetch(/api/heatmap/avg_temp?seasonsummeryear2023) .then(response response.json()) .then(heatData { L.heatLayer(heatData, { radius: 25, // 热力点半径 blur: 15, // 模糊度 maxZoom: 10, gradient: {0.4: blue, 0.6: lime, 0.8: yellow, 1.0: red} // 颜色梯度 }).addTo(map); });后端需要提供一个专门的接口来处理这种聚合查询例如使用SQL的AVG()函数和GROUP BY子句来计算每个城市的夏季平均温度并连同经纬度一起返回。实操心得前端性能优化当历史数据量很大时比如展示一整年的每日数据直接返回所有原始点给前端绘制折线图可能会导致浏览器卡顿。有两种优化策略一是后端进行数据采样例如对于超过365个点的数据计算其每周或每月的平均值后再返回二是利用ECharts的数据采样sampling功能。对于热力图如果城市点很多也需要适当调整radius和blur参数并在后端进行必要的数据聚合避免前端计算压力过大。5. 系统部署与运维实践5.1 项目结构与依赖管理一个清晰的项目结构是长期维护的基础。我的项目目录大致如下weather_system/ ├── app.py # Flask应用主入口 ├── config.py # 配置文件开发/生产环境 ├── requirements.txt # Python依赖列表 ├── run_scheduler.py # 独立运行的定时任务脚本 ├── /scrapers # 爬虫模块目录 │ ├── __init__.py │ ├── base_spider.py # 基础爬虫类 │ ├── realtime_spider.py │ └── historical_spider.py ├── /models # 数据库模型 │ ├── __init__.py │ └── weather_models.py ├── /routes # Flask蓝本/路由 │ ├── __init__.py │ ├── api.py # 数据API │ └── views.py # 页面路由 ├── /static # 静态资源 │ ├── /js │ ├── /css │ └── /geojson # 存放山东地图GeoJSON文件 ├── /templates # Jinja2模板 │ └── index.html └── /logs # 日志目录使用requirements.txt固化依赖版本Flask2.3.3 Flask-SQLAlchemy3.0.5 Flask-CORS4.0.0 APScheduler3.10.4 requests2.31.0 beautifulsoup44.12.2 lxml4.9.35.2 定时任务管理APScheduler数据爬取需要定时执行。我选择使用APScheduler并将其集成在Flask应用上下文中。# 在app.py或专门的scheduler模块中 from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from scrapers.realtime_spider import run_realtime_spider from scrapers.historical_spider import run_historical_spider scheduler BackgroundScheduler() # 每30分钟执行一次实时爬虫 scheduler.add_job( funcrun_realtime_spider, triggerCronTrigger(minute*/30), idrealtime_job, replace_existingTrue ) # 每天凌晨2点执行历史数据爬虫补全前一天的数据 scheduler.add_job( funcrun_historical_spider, triggerCronTrigger(hour2, minute0), idhistorical_job, replace_existingTrue ) # 在Flask应用工厂函数中启动调度器 def create_app(): app Flask(__name__) # ... 其他配置 ... if not app.debug or os.environ.get(WERKZEUG_RUN_MAIN) true: scheduler.start() # 优雅关闭 atexit.register(lambda: scheduler.shutdown()) return app注意生产环境部署在开发服务器如Flask内置服务器中运行APScheduler通常没问题。但在使用Gunicorn等WSGI服务器部署生产环境时需要特别注意。Gunicorn会启动多个工作进程可能导致定时任务被重复启动。解决方案通常是将定时任务作为一个独立的进程运行例如使用Celery或者确保只在Gunicorn的Master进程中启动调度器这需要特定的配置。5.3 基础部署方案以Linux Nginx Gunicorn为例服务器准备购买一台云服务器如腾讯云、阿里云ECS安装Ubuntu系统。环境配置通过SSH登录服务器安装Python3、pip、虚拟环境venv、Nginx和Git。sudo apt update sudo apt install python3-pip python3-venv nginx git -y拉取代码并安装依赖cd /var/www sudo git clone 你的项目仓库地址 weather_system cd weather_system python3 -m venv venv source venv/bin/activate pip install -r requirements.txt配置Gunicorn创建Gunicorn服务文件。sudo vim /etc/systemd/system/weather_system.service内容如下[Unit] DescriptionGunicorn instance for Shandong Weather System Afternetwork.target [Service] Userwww-data Groupwww-data WorkingDirectory/var/www/weather_system EnvironmentPATH/var/www/weather_system/venv/bin ExecStart/var/www/weather_system/venv/bin/gunicorn --workers 3 --bind unix:weather_system.sock -m 007 app:create_app() [Install] WantedBymulti-user.target启动并启用服务sudo systemctl start weather_system sudo systemctl enable weather_system配置Nginxsudo vim /etc/nginx/sites-available/weather_system内容如下server { listen 80; server_name your_domain.com; # 替换为你的域名或服务器IP location / { include proxy_params; proxy_pass http://unix:/var/www/weather_system/weather_system.sock; } # 静态文件交由Nginx直接处理效率更高 location /static { alias /var/www/weather_system/static; expires 30d; } }创建软链接并重启Nginxsudo ln -s /etc/nginx/sites-available/weather_system /etc/nginx/sites-enabled sudo nginx -t sudo systemctl restart nginx处理定时任务如前所述在生产环境中更稳妥的方式是将爬虫脚本设置为独立的系统Cron Job或者使用像supervisor这样的进程管理工具来运行一个独立的Python脚本即run_scheduler.py而不是将其放在Web应用进程中。6. 常见问题排查与优化技巧6.1 爬虫常见问题返回数据为空或解析失败原因网站改版HTML结构变化请求被反爬机制拦截如IP被封、请求头不完整。排查首先打印resp.text的前几百字符看是否返回了预期的HTML还是反爬提示如验证码页面。使用浏览器开发者工具对比你的爬虫请求和正常浏览器请求的Headers差异特别是User-Agent,Cookie,Referer。解决更新解析逻辑完善请求头添加随机延迟考虑使用IP代理池对于大规模爬取。数据库写入错误如唯一约束冲突原因试图插入(city_id, date)重复的数据。解决在插入前使用db.session.merge()方法它会自动判断是插入还是更新。或者采用“先查询存在则更新不存在则插入”的逻辑。record HistoricalWeather.query.filter_by(city_idcity_id, datetarget_date).first() if record: record.high_temp new_high_temp # ... 更新其他字段 else: new_record HistoricalWeather(city_idcity_id, datetarget_date, ...) db.session.add(new_record) db.session.commit()6.2 前端可视化问题地图或图表不显示原因静态资源路径错误JavaScript库未正确加载API接口返回数据格式不符合ECharts/Leaflet要求。排查打开浏览器控制台F12查看“网络(Network)”标签页确认JS/CSS文件、API请求是否成功加载状态码200并检查API返回的JSON数据结构。解决修正静态文件URL确保API返回的数据是前端库期望的格式如数组、对象。页面加载缓慢原因一次性请求数据量过大前端图表渲染大量数据点未启用Gzip压缩。解决后端实现数据分页或聚合前端利用ECharts的dataZoom组件或采样功能在Nginx中开启Gzip压缩。gzip on; gzip_vary on; gzip_min_length 1024; gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xmlrss text/javascript;6.3 系统性能优化建议数据库层面定期归档实时天气表数据增长极快可以定期如每月将旧数据迁移到历史归档表保持主表轻量。查询优化对复杂查询使用EXPLAIN命令分析执行计划确保索引被正确使用。避免在循环中执行数据库查询N1问题使用joinedload等 eager loading 技术。连接池在生产环境中配置SQLAlchemy的连接池参数避免频繁创建销毁数据库连接。缓存策略对于变化不频繁的数据如城市列表、某城市过去30天的历史数据概览可以使用Redis或Memcached进行缓存。Flask有Flask-Caching扩展可以方便地集成。from flask_caching import Cache cache Cache(config{CACHE_TYPE: SimpleCache}) # 开发环境用简单缓存 cache.init_app(app) app.route(/api/city_list) cache.cached(timeout3600) # 缓存1小时 def get_city_list(): cities City.query.all() # ... 序列化并返回异步处理如果爬虫任务非常耗时如爬取全省所有城市多年历史数据可以考虑使用Celery等异步任务队列将任务丢到后台执行避免阻塞Web请求。前端可以通过轮询或WebSocket来获取任务进度和结果。这个项目从构思到实现踩了不少坑但也收获满满。最大的体会是一个完整的数据系统从数据获取、清洗、存储到最终的可视化呈现每一个环节都需要仔细设计和反复调试。尤其是面对网络爬虫的不确定性健壮的错误处理和日志记录是保证系统长期稳定运行的关键。现在我可以随时打开这个自己搭建的天气系统一眼看清齐鲁大地的风云变幻这种掌控数据的感觉正是编程的乐趣所在。如果你也想尝试不妨从爬取一个城市的数据开始逐步扩展过程中遇到问题善用搜索引擎和社区大部分难题都能找到解决方案。本文还有配套的精品资源点击获取