四件套搭建爬虫数据可视化大屏：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, bodyhtml, encodingutf-8, requestrequest ) return None这里return HtmlResponse会直接结束下载流程把渲染后的内容送进回调函数中间件本身没有去请求网络。注意用这个方案时Spider对象的playwright_page_grab方法里要维护一个浏览器实例不要为每个请求都新建浏览器否则内存占用会很大。常规的设置是每个Spider只创建一个playwright异步上下文用信号量控制并发数量。实践中我一般不建议对列表页也启用JS渲染。只有确认iframe内部是真正需要的字段时才用这种方案。能用请求拼接解决的动态内容一律不用浏览器。2.3 反爬与运行节奏Cookie、延时和并发Scrapy默认的并发和下载速度非常快快到一个小时就能把对方服务器打满。大屏项目的数据量通常不大完全没必要追求极速。我在settings.py里一般这样配置参数推荐值说明ROBOTSTXT_OBEYFalse大屏抓取通常不需要遵守robots但注释掉也要自己控制频率DOWNLOAD_DELAY1.5每次请求间隔1.5秒观察对方限流阈值再下调CONCURRENT_REQUESTS8并发数越大越容易被封压测时从16降到8COOKIES_ENABLEDTrue有些站点的登录态判断依赖Cookie开启后注意保存DEFAULT_REQUEST_HEADERS见代码补全UA和Referer# 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) ) ENGINEInnoDB;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(dictionaryTrue) 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要稳。dictionaryTrue让游标返回字典而不是元组这样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_namemypool, pool_size5, host127.0.0.1, userroot, passwordyourpass, databasedashboard ) def get_db(): return pool.get_connection()pool_size5表示最多同时给5个请求分配连接超过的请求等待。大屏页面一般同时发起四个左右的接口请求5个连接刚好够用。如果数据量变大需要设置pool_reset_sessionTrue否则连接复用后会话变量可能脏数据。5. 用ECharts拼大屏地图、饼图、折线图的联动实现5.1 大屏页面骨架静态页面加载EChartsECharts大屏最稳妥的写法是写一个静态HTML页面用script标签加载ECharts通过fetch从Flask接口拿数据再调用setOption渲染图表。这个方案对服务端压力小也便于在本地直接打开HTML调试样式。!DOCTYPE html html langzh-CN head meta charsetUTF-8 title数据可视化大屏/title script srchttps://cdn.jsdelivr.net/npm/echarts5.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 idmap/div div idpie/div div idtrend/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.showfalse可以在鼠标悬浮时隐藏城市名避免一片密密麻麻的文字。还可以在地图上叠加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和valuefetch(/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 ConfigurationScript路径填scrapy的cmdline.py参数填crawl product。最后一个小技巧给Flask接口加一个简单的缓存头大屏上几十个图表同时在拉数据时MySQL连接池会被瞬间占满。在路由响应上加bp.after_request设置Cache-Control: max-age60就能让同一分钟内重复打开的浏览器直接读取本地缓存大幅降低数据库压力。bp.after_request def add_header(resp): resp.headers[Cache-Control] public, max-age60 return resp把这个缓存挂在地图接口和汇总接口上大屏首次加载可能慢半秒后续切换页面几乎无延迟。数据定时更新后缓存自然过期等下一轮抓取就有新数据可看。本文还有配套的精品资源点击获取