四件套搭建爬虫数据可视化大屏：Scrapy、MySQL、Flask与ECharts实战

简介：这是一套基于Echarts、Flask、Scrapy和MySQL的Python爬虫与数据可视化大屏设计源码，面向希望构建完整数据采集-存储-展示链路的数据分析开发者，适合用于学习爬虫框架、Web后端和前端图表的组合实践。压缩包共61个文件，约6.05MB，包含13个Python脚本、17个XML配置、7个JavaScript文件、SQL脚本、HTML模板及图片样式等，其中Python脚本覆盖爬虫与数据入库逻辑，JS与HTML实现大屏可视化，XML与配置文件则用于项目环境设定。已有793人学习下载，是一份被多次查阅的实战参考资料。通过源码可以掌握Scrapy抓取网页数据并写入MySQL、Flask搭建后端接口、Echarts动态渲染大屏图表的完整流程，还可参考其目录结构与配置方式，快速迁移到自己的数据分析项目中。

1. 一套四件套讲清楚的事：爬虫与数据可视化大屏

今天早上一个指标变了，昨天还排在前面的那条记录今天可能掉到第二页。你想盯住它，最省事的办法是写个爬虫定时抓，把结果存进数据库，再用一个页面把变化画出来。基于echarts、Flask、Scrapy和MySQL的这套组合，就是解决这个问题的常见结构：Scrapy负责采集，MySQL负责存储，Flask负责把数据变成接口，ECharts负责把接口变成大屏。四个组件各管一段，边界清晰，出了问题能快速定位在哪一层，这也是大量开源“爬虫大屏”项目的标准骨架。如果你有Python基础、能在服务器上跑定时任务，后面这些内容可以让你从空目录开始，把一条可运行的数据链路完整拼出来。

2. 用Scrapy采集：Item设计、动态页面与反爬应对

2.1 从页面结构确定Item字段和Spider回调

Scrapy的工作方式不是先写数据库再写爬虫，而是先从HTML里找到你要的字段，定义成Item，再写Spider去填充。我一般先打开目标页面，用浏览器开发者工具看列表区里每条数据的DOM特征，确认是静态渲染还是动态加载。静态页面直接用CSS选择器就能取到值，动态页面则要另想办法。下面是一个典型的商品列表Item定义：

# items.py import scrapy class ProductItem(scrapy.Item): # 业务主键，用于后续MySQL去重 biz_id = scrapy.Field() name = scrapy.Field() price = scrapy.Field() source_url = scrapy.Field() updated_at = scrapy.Field()

字段名称和数据库列名尽量保持一致，这样后面写Pipeline时不用做大量重命名。biz_id是这条记录在来源网站上唯一标识，没有它，去重就只能靠名字+价格组合，碰到价格变动就会插入重复数据。对应的Spider写法是把选择器结果循环，逐条填进Item然后yield：

# spiders/product.py import scrapy from scraper.items import ProductItem class ProductSpider(scrapy.Spider): name = 'product' start_urls = ['https://example.com/catalog'] def parse(self, response): for card in response.css('div.product-card'): item = ProductItem() item['biz_id'] = card.css('::attr(data-id)').get() item['name'] = card.css('.title::text').get() item['price'] = card.css('.price::text').get() item['source_url'] = response.url item['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') yield item

这里的::attr(data-id)是取># middlewares.py class PlaywrightMiddleware: def process_request(self, request, spider): if request.meta.get('render_js'): # 交给Playwright执行JS后，返回新响应 html = spider.playwright_page_grab(request.url) return HtmlResponse( request.url, body=html, encoding='utf-8', request=request ) return None

这里return HtmlResponse会直接结束下载流程，把渲染后的内容送进回调函数，中间件本身没有去请求网络。注意用这个方案时，Spider对象的playwright_page_grab方法里要维护一个浏览器实例，不要为每个请求都新建浏览器，否则内存占用会很大。常规的设置是每个Spider只创建一个playwright异步上下文，用信号量控制并发数量。

实践中我一般不建议对列表页也启用JS渲染。只有确认iframe内部是真正需要的字段时，才用这种方案。能用请求拼接解决的动态内容，一律不用浏览器。

2.3 反爬与运行节奏：Cookie、延时和并发

Scrapy默认的并发和下载速度非常快，快到一个小时就能把对方服务器打满。大屏项目的数据量通常不大，完全没必要追求极速。我在settings.py里一般这样配置：

# settings.py BOT_NAME = 'scraper' SPIDER_MODULES = ['scraper.spiders'] ROBOTSTXT_OBEY = False DOWNLOAD_DELAY = 1.5 CONCURRENT_REQUESTS = 8 COOKIES_ENABLED = True DEFAULT_REQUEST_HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Referer': 'https://example.com', }

DOWNLOAD_DELAY设置过小会导致请求集中在几秒内发出，对方日志里会看到异常尖峰。大屏数据一天刷一次就够的场景，把间隔设置在2到5秒都行。碰到403或验证码时，优先检查有没有漏掉某个请求头，其次才是考虑Cookie失效，因为大部分反爬拦截都发生在Headers上。

3. 用MySQL落库：表结构设计、去重与增量更新

3.1 表结构设计：一张主表够用，别急着分库

爬虫量级在几十万行以内时，一张业务主表加必要索引就能满足大屏查询。多次重造轮子的经验是，在开始写Pipeline之前先建好表，字段类型直接决定后续索引和查询方案。名称存VARCHAR，价格用DECIMAL，时间用DATETIME，这比全都用VARCHAR省一半空间，排序也不会出错。

CREATE DATABASE dashboard DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE TABLE product ( id INT PRIMARY KEY AUTO_INCREMENT, biz_id VARCHAR(64) NOT NULL UNIQUE, name VARCHAR(255) NOT NULL, price DECIMAL(10,2), source_url VARCHAR(512), created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_updated_at (updated_at), KEY idx_name (name) ) ENGINE=InnoDB;

utf8mb4一定要选，因为utf8在MySQL里是utf8mb3，存不了4字节的Emoji和部分生僻字。爬虫抓到的标题里只要出现一个特殊符号，写入就报1366错误。biz_id上的UNIQUE约束是去重的最后一道保险，即使Pipeline逻辑写错了，数据库这层也会把重复记录挡下来。updated_at上建索引，是因为后面做增量查询时最常用的条件就是这个字段。

3.2 MySQL Pipeline：去重、更新与批量提交

Scrapy的Pipeline在Item通过后逐个执行，每个Pipeline只负责一件事。我把MySQL写入独立成一个Pipeline，里面同时处理连接、去重和批量提交：

# pipelines.py import mysql.connector class MysqlPipeline: def __init__(self, mysql_config): self.mysql_config = mysql_config @classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get('MYSQL_CONFIG')) def open_spider(self, spider): self.conn = mysql.connector.connect(**self.mysql_config) self.cursor = self.conn.cursor() self.buffer = [] def process_item(self, item, spider): self.buffer.append(dict(item)) if len(self.buffer) >= 500: self._flush() return item def close_spider(self, spider): self._flush() self.cursor.close() self.conn.close() def _flush(self): sql = """ INSERT INTO product (biz_id, name, price, source_url, updated_at) VALUES (%(biz_id)s, %(name)s, %(price)s, %(source_url)s, %(updated_at)s) ON DUPLICATE KEY UPDATE name = VALUES(name), price = VALUES(price), updated_at = VALUES(updated_at) """ self.cursor.executemany(sql, self.buffer) self.conn.commit() self.buffer.clear()

open_spider里只建立一次连接，避免每条Item都重复握手。executemany加commit放在同一个_flush里，每500条提交一次事务，比每条都commit快一个数量级。ON DUPLICATE KEY UPDATE的含义是：当biz_id冲突时，更新名称、价格和更新时间，而不是报错，这正好解决“同一商品今天抓到新价格”的更新需求。

3.3 增量更新：用updated_at跑每日增量

大屏上展示的“今日趋势”需要知道当天有多少条数据更新过。与其全表扫描，不如让爬虫只输出当天改动的字段，然后依靠上面的ON DUPLICATE KEY UPDATE做更新。Flask端查询时用updated_at >= 当天零点即可拿到当天变化集合。

SELECT COUNT(*) AS total, DATE(updated_at) AS d FROM product WHERE updated_at >= '2025-01-01 00:00:00' GROUP BY DATE(updated_at) ORDER BY d;

DATE(updated_at)把时间截断到天，可以当作折线图x轴。注意如果表数据量超过百万，这个查询要在updated_at索引支持下运行，否则全表扫描会拖慢接口响应。还有一点，ON DUPLICATE KEY UPDATE里如果updated_at不更新，那么重复抓到的数据时间不会变，当天的统计就会缺失，所以更新的字段里一定要包含时间值。

4. 用Flask出接口：蓝图、连接池与JSON规范

4.1 为什么用Blueprint：把大屏接口按模块分隔

项目只要超过三个接口，Flask单文件就开始难维护。用Blueprint把接口按业务拆开，主应用只负责注册，每个模块自己管自己的路由和查询逻辑。这种做法不会增加运行开销，但对后续加接口、加权限都很友好。

# dashboard/views.py from flask import Blueprint, jsonify dashboard_bp = Blueprint('dashboard', __name__) @dashboard_bp.route('/api/summary') def summary(): # 汇总卡片数据 return jsonify({'code': 0, 'data': {'total': 12345}})

Blueprint创建后，需要在主应用的app.py里注册：

# app.py from flask import Flask from dashboard.views import dashboard_bp app = Flask(__name__) app.register_blueprint(dashboard_bp)

register_blueprint默认挂在根路径下，如果以后想把整个模块迁移到/api/v2，只需要改一行前缀参数。接口返回格式统一为JSON对象，包含code和data两个字段，前端拿到code不等于0时弹出错误信息，不用再解析各种异常结构。

4.2 接口定义：summary、map、trend三个路由

大屏页面通常要三类数据：汇总卡片、地图分布、趋势折线。对应到Flask就是三个路由，每个路由里只写与自身相关的查询语句：

from flask import Blueprint, jsonify, request from db import get_db bp = Blueprint('api', __name__, url_prefix='/api') @bp.route('/summary') def summary(): db = get_db() cursor = db.cursor(dictionary=True) cursor.execute("SELECT COUNT(*) AS total FROM product") total = cursor.fetchone()['total'] cursor.execute(""" SELECT DATE(updated_at) AS d, COUNT(*) AS cnt FROM product WHERE updated_at >= %s GROUP BY DATE(updated_at) ORDER BY d """, (request.args.get('start', '2025-01-01'),)) trend = cursor.fetchall() return jsonify({'code': 0, 'data': {'total': total, 'trend': trend}}) @bp.route('/map') def map_data(): # 返回省/地区维度的聚合数据 return jsonify({'code': 0, 'data': []})

db = get_db()里的get_db来自一个公共模块，负责从连接池拿连接。每次请求自己开连接、自己关闭，用完放回池里，比在路由里反复mysql.connector.connect要稳。dictionary=True让游标返回字典而不是元组，这样jsonify能直接把结果序列化。注意%s占位符传参来自前端时要用参数名字典，不要用字符串拼接，否则SQL注入风险很高。

4.3 连接管理与CORS设置

大屏页面如果和Flask不在同一域名下，浏览器会拦截跨域请求。最常见的处理是用flask-cors扩展：

from flask_cors import CORS app = Flask(__name__) CORS(app, resources={r"/api/*": {"origins": "*"}})

origins: "*"表示允许所有来源，适合内网大屏；公网部署建议改成具体域名。Flask和ECharts在同一台服务器上时，即使没有跨域问题，也建议加上CORS配置，后面调试时可以用curl直接从本地访问接口，省去代理配置的麻烦。

连接管理方面，我推荐用mysql.connector.connect配固定连接池：

import mysql.connector from mysql.connector import pooling pool = pooling.MySQLConnectionPool( pool_name='mypool', pool_size=5, host='127.0.0.1', user='root', password='yourpass', database='dashboard' ) def get_db(): return pool.get_connection()

pool_size=5表示最多同时给5个请求分配连接，超过的请求等待。大屏页面一般同时发起四个左右的接口请求，5个连接刚好够用。如果数据量变大，需要设置pool_reset_session=True，否则连接复用后会话变量可能脏数据。

5. 用ECharts拼大屏：地图、饼图、折线图的联动实现

5.1 大屏页面骨架：静态页面加载ECharts

ECharts大屏最稳妥的写法是写一个静态HTML页面，用<script>标签加载ECharts，通过fetch从Flask接口拿数据，再调用setOption渲染图表。这个方案对服务端压力小，也便于在本地直接打开HTML调试样式。

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>数据可视化大屏</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <style> #map { width: 40%; height: 400px; float: left; } #pie { width: 30%; height: 400px; float: left; } #trend { width: 30%; height: 400px; float: left; } </style> </head> <body> <div id="map"></div> <div id="pie"></div> <div id="trend"></div> <script> // 下方详细加载逻辑 </script> </body> </html>

页面结构要提前占好容器，每个图表的div都要有明确宽度和高度，ECharts在容器尺寸为0时初始化会得到一团空白。我把三个图表放到一行，宽度用百分比分配，同时调用window.addEventListener('resize', () => { chart.resize(); })，这样窗口变化时图表能自适应。

5.2 中国地图与全局样式：地图JSON注册与配置

ECharts 5里地图数据不再内置，需要手动注册geojson。常见路径是把china.json放到static/目录下，在渲染前用fetch加载并注册。

fetch('static/china.json') .then(res => res.json()) .then(geo => { echarts.registerMap('china', geo); initMap(); });

注册完成后，initMap里写地图系列配置：

function initMap() { const chart = echarts.init(document.getElementById('map')); fetch('/api/map') .then(res => res.json()) .then(json => { chart.setOption({ title: { text: '全国数据分布', left: 'center' }, tooltip: {}, series: [{ type: 'map', map: 'china', roam: true, data: json.data, emphasis: { label: { show: false } } }] }); }); }

roam: true允许用户缩放和拖动地图，大屏演示时这个交互很受欢迎。地图针对的data格式是[{name: '北京', value: 23}, ...]，name必须与geojson里的城市名完全一致，写错一个字该区域就不显示数据。emphasis.label.show=false可以在鼠标悬浮时隐藏城市名，避免一片密密麻麻的文字。

还可以在地图上叠加markPoint标记重点城市：

series: [{ type: 'map', map: 'china', markPoint: { symbolSize: 10, data: [ { name: '北京', coord: [116.4, 39.9], value: 100 }, { name: '上海', coord: [121.47, 31.23], value: 80 } ] } }]

coord是经纬度坐标，markPoint会直接把标记画在地图对应位置，与数据中是否包含该城市无关。从接口返回的图标数据即使没有北京这一条，只要坐标对，标记也能显示。

5.3 饼图、折线图与markPoint的联动：从JSON到图表的映射

饼图和折线图的数据结构不同，但逻辑与地图完全一致，都是等接口返回后setOption。饼图最常用的配置是给一个数组，每个元素有name和value：

fetch('/api/trend') .then(res => res.json()) .then(json => { const pieChart = echarts.init(document.getElementById('pie')); pieChart.setOption({ series: [{ type: 'pie', radius: ['45%', '70%'], roseType: 'radius', data: json.data }] }); const trendChart = echarts.init(document.getElementById('trend')); trendChart.setOption({ xAxis: { type: 'category', data: json.trend.map(d => d.d) }, yAxis: { type: 'value' }, series: [{ type: 'line', data: json.trend.map(d => d.cnt), smooth: true }] }); });

radius: ['45%', '70%']表示环形饼图，中间空的区域正好可以放大屏的标题或核心数字。roseType: 'radius'是南丁格尔玫瑰图，让每块扇形的半径按值缩放，视觉效果更夸张。折线图用xAxis放日期、yAxis放数量，smooth: true让曲线带圆角，看起来更流畅。

ECharts的坐标轴数据不能直接吃接口返回的对象，必须用map提取字段。如果接口返回的是[{d: '2025-01-01', cnt: 10}]，直接data: json.trend会渲染不出任何东西，必须转成[10, 20, 30]这种数组。这个转化逻辑我一般放在fetch之后、setOption之前，方便出错时打印对比。

6. 上线后的小技巧：用空响应排查问题

大屏做出来之后，真正费时间的是排查“明明数据库有数，图表就是空白”这一类问题。我总结了一套从后往前查的固定路径，配合几个小技巧能快速定位。

先从Flask接口开始验证：

curl "http://127.0.0.1:5000/api/summary"

如果返回的不是JSON，而是HTML或500错误页，说明问题出在你的路由代码里，而不是前端。如果接口正常返回JSON但页面仍然空，就在浏览器控制台里打印json.data，看数据是否为空数组。空数组通常意味着SQL查询的过滤条件太严格，比如updated_at的日期范围写的晚于数据最晚时间，自然查不到任何记录。

一个非常常见又隐蔽的问题：地图数据里的城市名与geojson里的名称不一致。比如geojson写的是“广西壮族自治区”，而接口返回的是“广西”，地图点选不上，但控制台不报错。我的排查方法是写个小函数把接口返回的名称和geojson里所有名称打印出来对照，再决定在SQL里统一别名，还是在接口层做一次映射。

另一个高频坑是ECharts图表渲染前容器宽度为0，用无头浏览器截图时经常遇到。解决方式是初始化图表前用document.getElementById('map').clientWidth检查宽度，如果为0，先延时再初始化，或者强制给容器加一个宽度。PyCharm里跑Scrapy时，如果直接右键运行product.py，会因为缺少scrapy命令入口而报错，正确的做法是配置一个Run Configuration，Script路径填scrapy的cmdline.py，参数填crawl product。

最后一个小技巧：给Flask接口加一个简单的缓存头，大屏上几十个图表同时在拉数据时，MySQL连接池会被瞬间占满。在路由响应上加@bp.after_request设置Cache-Control: max-age=60，就能让同一分钟内重复打开的浏览器直接读取本地缓存，大幅降低数据库压力。

@bp.after_request def add_header(resp): resp.headers['Cache-Control'] = 'public, max-age=60' return resp

把这个缓存挂在地图接口和汇总接口上，大屏首次加载可能慢半秒，后续切换页面几乎无延迟。数据定时更新后，缓存自然过期，等下一轮抓取就有新数据可看。

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