基于Python Flask与ECharts的山东天气数据采集与可视化系统实战

简介：数据采集与可视化是数据分析领域的基础环节，其核心原理是通过技术手段从互联网等数据源获取信息，并转化为直观的图表进行洞察。在Web开发实践中，Python因其丰富的库生态成为实现这一流程的利器，结合轻量级框架能快速构建数据处理应用。本文聚焦于一个具体的技术实现：使用Python的Flask框架搭建后端服务，通过requests和BeautifulSoup库构建稳健的爬虫，从公开数据源采集山东省各地市的实时与历史天气数据，并利用ECharts库进行多维度可视化呈现。该项目完整覆盖了从数据采集、清洗存储、API接口设计到前端图表渲染的工程链路，为学习Python全栈开发、理解时间序列数据处理及应对反爬策略提供了典型的实战案例。

1. 项目概述与核心价值

最近在做一个挺有意思的私人项目，起因是想看看老家山东这几年的天气变化趋势，顺便练练手。这个项目说白了，就是一个能自动抓取山东省各地市实时和历史天气数据，然后存起来，最后用各种图表直观展示出来的Web系统。听起来好像挺简单，但真做起来，从数据源的稳定性、爬虫策略的制定，到后端数据存储、前端可视化呈现，每一步都有不少门道。我选择了Python作为主力语言，用Flask搭建Web后端，数据可视化则交给了ECharts和Pyecharts。整个过程下来，感觉像完成了一次小型的数据工程全栈实践，对数据处理流程的理解深了不少。

这个系统主要解决了几个痛点：一是市面上现成的天气历史数据查询要么收费，要么粒度不够细；二是想自己分析某个地区的长期气象规律时，缺乏一个集数据获取、存储、分析、展示于一体的工具。通过这个项目，你可以一站式地看到山东省内任意城市的温度、湿度、风速、降水等关键气象要素的实时状态、历史走势和空间分布。无论是想研究气候变化，还是单纯想为出行做个参考，这个工具都能派上用场。对于正在学习Python Web开发、数据爬虫或者数据可视化的朋友来说，这个项目涵盖了从数据采集到应用呈现的完整链路，是个非常不错的练手素材。

2. 系统整体架构与技术选型

2.1 核心架构设计思路

整个系统的设计遵循了典型的数据处理流水线：数据采集 -> 数据清洗与存储 -> 数据服务与接口 -> 数据可视化展示。架构上采用了前后端分离的轻量级模式，后端用Flask提供RESTful API，前端主要使用HTML/CSS/JS配合ECharts库来渲染图表，两者通过Ajax进行数据交互。这样设计的好处是前后端职责清晰，后端专注于数据处理和业务逻辑，前端专注于用户交互和视图渲染，也便于后续的维护和扩展。

为什么选择Flask而不是Django？对于这个规模的项目，Flask的轻量、灵活和“微”框架特性更符合需求。我们不需要Django自带的那一整套庞大的Admin、ORM（虽然也可以用Flask-SQLAlchemy）和模板系统，Flask允许我们从零开始按需组装，对理解Web框架底层原理更有帮助。数据存储方面，考虑到天气数据是典型的时间序列数据，且需要支持城市维度的查询，选择了关系型数据库MySQL，并设计了合理的索引来优化查询速度。对于需要快速检索的配置信息（如城市代码映射），也可以辅以Redis作为缓存。

2.2 关键技术组件解析

1. 数据爬取层（Scraper）：这是系统的数据源头。核心工具是requests库用于发送HTTP请求，BeautifulSoup4或lxml用于解析HTML页面。考虑到目标网站可能有反爬机制，需要合理设置请求头（User-Agent、Referer等）、使用time.sleep()进行礼貌爬取，并可能需要处理IP被封的问题（对于个人项目，使用代理池可能过于复杂，更可行的是降低请求频率和寻找更友好的数据接口）。
2. 数据存储层（Storage）：使用SQLAlchemy作为ORM工具，它与Flask集成良好（Flask-SQLAlchemy）。数据库表设计是关键，至少需要两张核心表：一张存储城市基本信息（城市ID、名称、行政区划代码、经纬度），另一张存储天气数据（关联城市ID、日期时间、温度、湿度、风速、降水量、天气现象等字段）。为日期和城市ID建立复合索引能极大提升按城市查询历史数据的效率。
3. Web服务层（Web Service）：Flask框架担当此任。需要编写多个API端点，例如：/api/current/<city_code>用于获取指定城市实时天气，/api/historical?city_code=xxx&start_date=xxx&end_date=xxx用于查询历史数据，/api/cities用于获取所有支持的城市列表。使用Flask-CORS处理可能的前端跨域请求。
4. 数据可视化层（Visualization）：这是系统的门面。选择百度开源的ECharts，功能强大且文档齐全。在Python后端，可以使用pyecharts库生成图表配置，然后在前端用ECharts渲染。也可以直接在前端通过Ajax调用后端API获取数据，然后用JavaScript初始化ECharts实例。我们将实现折线图（趋势分析）、柱状图（要素对比）、地图热力图（空间分布）等多种图表类型。

注意：在爬取公开天气数据时，务必遵守网站的robots.txt协议，控制请求速率，避免对目标服务器造成压力。本项目旨在技术学习，所爬取的数据应为个人、非商业用途。

3. 核心模块实现细节

3.1 天气数据爬虫的稳健性设计

数据爬取是整个系统的基石，其稳健性直接决定了数据的质量和系统的可靠性。我们的目标是爬取山东省下辖16个地级市的天气数据。

第一步：确定数据源与解析策略经过调研，可以选择中国天气网等提供公开数据的网站。首先需要分析其页面结构。以济南为例，找到包含实时天气信息的页面元素。通常，温度、湿度、风速等信息会放在特定的HTML标签内，通过CSS类或ID标识。使用浏览器的开发者工具（F12）的“检查”功能，可以定位到这些元素。然后，在Python爬虫中，使用BeautifulSoup或lxml配合相应的选择器（如CSS选择器或XPath）将这些信息提取出来。

例如，获取温度的可能代码片段：

import requests from bs4 import BeautifulSoup def fetch_current_weather(city_code): url = f"https://www.weather.com.cn/weather/{city_code}.shtml" headers = {'User-Agent': '你的浏览器User-Agent'} try: resp = requests.get(url, headers=headers, timeout=10) resp.encoding = 'utf-8' soup = BeautifulSoup(resp.text, 'lxml') # 假设温度在class为‘tem’的span标签里 temp_tag = soup.find('span', class_='tem') temperature = temp_tag.text.strip() if temp_tag else None # 类似地提取湿度、风速等... return {'temperature': temperature, ...} except requests.RequestException as e: print(f"爬取{city_code}数据失败: {e}") return None

第二步：构建城市代码映射与定时任务需要建立一个山东省城市名称与其在数据源网站上对应代码的映射字典。这个映射需要手动从网站页面或网络请求中分析得出。爬虫脚本需要遍历这个字典，依次抓取每个城市的数据。

为了实现数据的实时性，需要让爬虫定时运行。在Linux服务器上，最直接的方式是使用crontab定时任务。例如，设置每30分钟运行一次爬虫脚本。在脚本内部，要做好错误处理（如网络超时、页面结构变更），并将成功抓取的数据立即存入数据库。更优雅的方式是使用APScheduler这样的Python库在Flask应用内部创建定时任务，但需要注意Flask的应用上下文管理。

第三步：历史数据抓取与增量更新历史数据的抓取更为复杂，因为通常需要按日期循环请求。有些网站提供历史数据页面，有些则需要分析其数据接口（API）。找到接口后，通过构造不同的日期参数进行请求。这里的关键是增量更新策略：每次抓取前，先查询数据库中该城市该日期是否已有数据，避免重复存储。同时，历史数据抓取频率可以低很多，比如每天一次，抓取前一天的数据，并需要处理可能的数据缺失或异常值。

3.2 数据存储与数据库设计

良好的数据库设计是高效查询的保障。我们使用Flask-SQLAlchemy来定义模型。

城市信息表 (City)：这张表是维度表，相对静态。

class City(db.Model): __tablename__ = 'city' id = db.Column(db.Integer, primary_key=True, autoincrement=True) code = db.Column(db.String(20), unique=True, nullable=False, comment='城市代码') name = db.Column(db.String(50), nullable=False, comment='城市名称') province = db.Column(db.String(20), default='山东', comment='所属省份') latitude = db.Column(db.Float, comment='纬度') longitude = db.Column(db.Float, comment='经度')

天气数据表 (WeatherData)：这是事实表，数据量会随时间增长。

class WeatherData(db.Model): __tablename__ = 'weather_data' id = db.Column(db.Integer, primary_key=True, autoincrement=True) city_id = db.Column(db.Integer, db.ForeignKey('city.id'), nullable=False, index=True) timestamp = db.Column(db.DateTime, nullable=False, index=True, comment='数据时间点') temperature = db.Column(db.Float, comment='温度(℃)') humidity = db.Column(db.Integer, comment='湿度(%)') wind_speed = db.Column(db.Float, comment='风速(m/s)') wind_direction = db.Column(db.String(10), comment='风向') precipitation = db.Column(db.Float, comment='降水量(mm)') weather_desc = db.Column(db.String(50), comment='天气现象') # 建立城市和时间的复合索引，这对按城市查历史数据至关重要 __table_args__ = ( db.Index('idx_city_timestamp', 'city_id', 'timestamp'), ) city = db.relationship('City', backref=db.backref('weather_records', lazy='dynamic'))

索引策略详解：

• 在city_id和timestamp上单独建立索引，可以加速基于单个条件的查询。
• 建立复合索引idx_city_timestamp，对于WHERE city_id = ? AND timestamp BETWEEN ? AND ?这类最常见的查询，其效率远高于两个单列索引。数据库可以直接在索引中定位到特定城市在某个时间范围内的数据位置，避免全表扫描。

数据清洗与入库：爬取到的原始数据往往需要清洗。例如，温度字符串可能包含“℃”符号，需要提取数字部分；风速可能是“3-4级”这样的文本，需要转换为具体的数值（如中间值3.5 m/s）。在入库前，编写一个清洗函数来处理这些情况，确保存入数据库的数据是干净、格式统一的。

3.3 Flask后端API开发

Flask应用的结构需要组织清晰。我推荐使用蓝图（Blueprint）来模块化路由。

应用工厂模式：使用应用工厂函数create_app()来创建Flask实例，这样便于配置管理和测试。

# app/__init__.py from flask import Flask from flask_sqlalchemy import SQLAlchemy from config import Config db = SQLAlchemy() def create_app(config_class=Config): app = Flask(__name__) app.config.from_object(config_class) db.init_app(app) # 注册蓝图 from app.api import bp as api_bp app.register_blueprint(api_bp, url_prefix='/api') return app

API蓝图与路由：在app/api目录下创建蓝图，并定义核心的数据接口。

# app/api/__init__.py from flask import Blueprint, jsonify, request from app.models import City, WeatherData from app import db bp = Blueprint('api', __name__) @bp.route('/cities', methods=['GET']) def get_cities(): """获取所有城市列表""" cities = City.query.all() return jsonify([{'code': c.code, 'name': c.name, 'lat': c.latitude, 'lon': c.longitude} for c in cities]) @bp.route('/current/<city_code>', methods=['GET']) def get_current_weather(city_code): """获取指定城市最新实时天气""" city = City.query.filter_by(code=city_code).first_or_404() # 获取该城市最新的一条记录 latest_data = WeatherData.query.filter_by(city_id=city.id).order_by(WeatherData.timestamp.desc()).first() if not latest_data: return jsonify({'error': 'No data found'}), 404 data = { 'city': city.name, 'timestamp': latest_data.timestamp.isoformat(), 'temperature': latest_data.temperature, 'humidity': latest_data.humidity, 'wind_speed': latest_data.wind_speed, 'precipitation': latest_data.precipitation, 'weather': latest_data.weather_desc } return jsonify(data) @bp.route('/historical', methods=['GET']) def get_historical_weather(): """查询历史天气数据""" city_code = request.args.get('city_code') start_date = request.args.get('start_date') # 格式：YYYY-MM-DD end_date = request.args.get('end_date') # 参数校验... city = City.query.filter_by(code=city_code).first_or_404() query = WeatherData.query.filter_by(city_id=city.id) if start_date: query = query.filter(WeatherData.timestamp >= start_date) if end_date: query = query.filter(WeatherData.timestamp <= end_date + ' 23:59:59') records = query.order_by(WeatherData.timestamp).all() # 将数据格式化为前端图表需要的格式，例如按日期分组求平均 # ... 数据处理逻辑 ... return jsonify(processed_data)

错误处理与API响应规范：使用Flask的@app.errorhandler装饰器统一处理404、500等错误，返回统一的JSON格式。对于API，良好的响应格式应包括状态码code、消息msg和实际数据data。

3.4 多维度数据可视化实现

可视化前端是一个独立的HTML页面，通过JavaScript调用后端API获取数据，并使用ECharts渲染。

地图热力图展示全省实时天气分布：这是最直观的展示方式。我们需要每个城市的实时温度和经纬度。

1. 前端通过/api/cities和循环调用/api/current/<code>（或后端提供一个批量接口）获取所有城市的实时数据。
2. 使用ECharts的geo组件和heatmap系列。将山东省的GeoJSON地图数据注册到ECharts（可以从公开的GIS数据网站获取）。
3. 将城市经纬度和温度值（或湿度、风速）组成[经度， 纬度， 数值]的数组，作为heatmap系列的data。
4. 通过视觉映射（visualMap）将数值映射到颜色梯度（如低温蓝色到高温红色），一张生动的全省气温分布热力图就生成了。

折线图与柱状图进行趋势与对比分析：

• 单城市多要素趋势图（折线图）：调用/api/historical接口，获取某个城市一段时间内的历史数据。在同一个直角坐标系（grid）下，使用多个line系列，分别绘制温度、湿度、风速的折线。由于量纲不同，需要用到ECharts的双Y轴（yAxis），左边轴对应温度（℃），右边轴对应湿度（%）或风速（m/s）。
• 多城市单要素对比图（柱状图）：比较同一天山东省内不同城市的某个气象要素。例如，比较各城市今日最高温。这需要后端提供一个聚合查询接口，返回指定日期各城市的某项数据（如日最高温）。前端使用bar系列，X轴为城市名称，Y轴为温度值，一目了然。

前端代码结构示例：

<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <title>山东天气可视化</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script> <script src="https://cdn.jsdelivr.net/npm/axios/dist/axios.min.js"></script> </head> <body> <div id="mapChart" style="width: 100%; height: 600px;"></div> <div id="lineChart" style="width: 100%; height: 400px;"></div> <script> // 初始化地图图表实例 var mapChart = echarts.init(document.getElementById('mapChart')); // 使用axios获取数据 axios.get('/api/cities').then(function (response) { var cities = response.data; // 进一步获取每个城市的实时数据，组装heatmap data... var heatmapData = []; cities.forEach(function(city){ axios.get('/api/current/' + city.code).then(function(resp){ var data = resp.data; heatmapData.push([city.lon, city.lat, data.temperature]); }); }); // 所有请求完成后，设置地图选项 var option = { title: { text: '山东省实时气温分布热力图' }, tooltip: {...}, visualMap: {...}, geo: { map: 'shandong', // 需提前注册'shandong'地图 ... // 其他配置 }, series: [{ type: 'heatmap', coordinateSystem: 'geo', data: heatmapData, pointSize: 10, blurSize: 5 }] }; mapChart.setOption(option); }); // 类似地初始化折线图... </script> </body> </html>

4. 部署、优化与问题排查

4.1 系统部署实践

开发完成后，需要将系统部署到服务器上，提供7x24小时的服务。我选择了一台Linux云服务器（如Ubuntu 20.04 LTS）。

1. 环境准备：在服务器上安装Python3、Pip、MySQL和Nginx。使用虚拟环境（venv）隔离项目依赖。
2. 代码部署：使用Git将代码克隆到服务器。在项目目录下安装依赖：pip install -r requirements.txt。
3. 数据库配置：在MySQL中创建数据库和用户，并导入表结构（可以通过Flask-Migrate生成迁移脚本并执行）。
4. 使用Gunicorn作为WSGI服务器：Flask自带的开发服务器不适合生产环境。使用Gunicorn来运行Flask应用。

   # 安装gunicorn pip install gunicorn # 启动应用，假设应用工厂函数在`app/__init__.py`中 gunicorn -w 4 -b 127.0.0.1:8000 "app:create_app()"

   -w 4表示启动4个工作进程，根据服务器CPU核心数调整。
5. 使用Nginx作为反向代理：Nginx处理静态文件、SSL加密，并将动态请求转发给Gunicorn。

   # 在Nginx站点配置中 server { listen 80; server_name your_domain.com; location / { proxy_pass http://127.0.0.1:8000; # 转发给Gunicorn proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } # 可添加静态文件处理规则 location /static { alias /path/to/your/app/static; } }

6. 进程管理：使用systemd或supervisor来管理Gunicorn进程，确保应用在崩溃或服务器重启后能自动恢复。

4.2 性能优化策略

随着数据量的增长，系统可能会变慢，可以从以下几个层面优化：

1. 数据库查询优化：
   • 确保索引有效：使用EXPLAIN语句分析慢查询，确保查询命中了我们建立的复合索引idx_city_timestamp。
   • 数据归档：天气历史数据具有很强的时间序列特性，对于非常久远的数据（如3年前），查询频率极低。可以定期将这类数据迁移到归档表，或者使用MySQL的分区表功能，按时间（如按月）进行分区，能显著提升近期数据的查询效率。
   • 查询字段精简：避免使用SELECT *，只查询前端需要的字段。
2. 应用层缓存：
   • 接口缓存：对于实时性要求不高的数据，如城市列表、昨日天气汇总等，可以使用Flask-Caching扩展，配合Redis或Memcached，对API响应进行缓存。例如，设置/api/cities接口缓存1天。
   • 页面片段缓存：对于复杂的可视化图表配置数据，如果数据本身更新不频繁，也可以将其序列化后缓存起来。
3. 前端优化：
   • 数据聚合与采样：当查询的时间范围非常大（如一年）时，一次性返回所有数据点（每天甚至每小时一个点）会导致数据量巨大，传输慢且前端渲染卡顿。后端应在查询时进行数据聚合，例如，查询一年数据时，按周或月返回平均值、最大值、最小值。或者，实现后端采样算法，在保证趋势不变的前提下减少数据点数量。
   • 懒加载与分页：对于历史数据表格展示，实现分页加载，而不是一次性加载所有数据。

4.3 常见问题与排查实录

在开发和运行过程中，我遇到了不少典型问题，这里记录下排查思路：

1. 爬虫突然获取不到数据，返回403或空页面

   • 可能原因：触发了网站的反爬机制（IP被封、请求头异常、访问频率过高）。
   • 排查：首先在浏览器中手动访问目标URL，确认页面正常。然后对比爬虫的请求头（特别是User-Agent、Cookie、Referer）与浏览器发出的请求头是否一致。使用requests的Session对象保持会话。最重要的，大幅降低请求频率，在请求间加入随机延时（如time.sleep(random.uniform(2, 5))）。
   • 解决：模拟浏览器请求头，添加合理的延时。考虑使用更友好的数据接口（如果有的话），而非直接爬取页面。

2. 地图热力图上城市位置不显示或错位

   • 可能原因：ECharts中注册的地图GeoJSON数据坐标系统与城市经纬度坐标系统不匹配；城市经纬度数据错误。
   • 排查：检查城市经纬度数据是否准确（可在地图软件上验证）。确认ECharts地图的坐标格式，通常是[经度, 纬度]，且经纬度是WGS84坐标系（即常见的GPS坐标）。
   • 解决：确保数据源中的经纬度格式正确。如果GeoJSON地图边界数据是另一种坐标系（如GCJ-02），需要对城市经纬度进行相应的坐标转换。

3. 历史数据查询接口响应缓慢

   • 可能原因：数据量太大；查询没有命中索引；存在N+1查询问题（例如，在序列化数据时，对每条记录都去查询关联的城市名称）。
   • 排查：打开数据库的慢查询日志。使用Flask-DebugToolbar等工具分析请求的SQL语句。检查是否使用了EXPLAIN来查看查询计划。
   • 解决：确保WHERE条件中的字段已建立索引。使用SQLAlchemy的joinedload或subqueryload来一次性加载关联对象，避免在循环中查询数据库。对于时间范围很长的查询，强制要求前端提供时间范围，并设置最大时间跨度限制（如不超过1年）。

4. 前端图表渲染大量数据时卡顿

   • 可能原因：ECharts实例中数据点过多（如超过数千个）。
   • 排查：检查从后端API返回的数据数组长度。
   • 解决：如前文所述，在后端进行数据聚合或采样。ECharts本身也提供了一些优化选项，如开启animation: false关闭动画，或使用large: true模式渲染大数据量。

5. 定时爬虫任务不执行或重复执行

   • 可能原因：服务器时间设置问题；crontab语法错误；脚本执行权限或环境变量问题；脚本本身有逻辑错误导致提前退出。
   • 排查：在crontab命令中，将输出重定向到日志文件（>> /path/to/log 2>&1），查看日志。手动在命令行下用完整路径和环境执行脚本，看是否成功。
   • 解决：在crontab中指定绝对路径和Python解释器路径。在脚本开头打印开始日志，结尾打印结束日志。对于更复杂的任务调度，考虑使用Celery替代crontab。

这个项目从构思到实现，再到优化，几乎踩遍了数据类应用常见的坑。但每解决一个问题，对系统设计的理解就加深一层。数据可视化不仅仅是把图表画出来，其背后稳定、高效、准确的数据流水线才是真正的价值所在。如果让我再优化一点，我会考虑引入消息队列（如Redis List或RabbitMQ）来解耦爬虫任务和数据入库过程，让系统的韧性和扩展性更强。

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