Python 爬虫项目 爬虫分库分表存储海量多品类采集数据

前言

随着爬虫业务持续扩张，采集范围从单一品类延伸至多行业、多渠道，数据量级也从万级、十万级逐步增长至百万、千万甚至亿级。传统单数据库、单数据表的存储架构会暴露出诸多瓶颈：单表数据量过大引发索引查询变慢、写入性能衰减、数据表锁竞争加剧；单库存储无法实现资源隔离，多品类数据混杂导致运维难度提升，一旦出现故障会影响全部业务数据。在此背景下，分库分表成为海量多品类爬虫数据落地存储的核心解决方案。

分库分表通过数据横向、纵向拆分，将海量数据分散至多个数据库与数据表中，有效降低单表、单库的数据压力，提升数据库读写性能、并发承载能力与系统可扩展性，同时结合业务品类做数据隔离，让不同赛道的爬虫数据独立运维、互不干扰。本文结合 Python 爬虫实战场景，围绕分库分表的设计思想、拆分规则、路由算法、代码实现、数据迁移、运维管理等内容展开深度讲解，区分垂直分库、水平分表、混合拆分三种主流架构，搭配完整可运行代码、底层原理剖析与生产环境落地规范，构建适配海量多品类爬虫的分布式存储体系。

项目所依赖的工具、第三方库官方资源链接整理如下，便于开发者查阅文档、部署环境：

1. PyMySQL：Python 操作 MySQL 数据库核心驱动，支持连接池、事务等高级特性
2. DBUtils：Python 数据库连接池组件，适配分库分表多连接场景
3. MySQL 官方文档：数据库配置、索引、事务、分区表相关参考资料
4. Python 官方下载地址：项目基础运行环境
5. Faker：模拟生成海量测试数据，用于分库分表压力测试

整套方案兼容单机多实例 MySQL、MySQL 集群环境，覆盖中小体量爬虫到分布式集群爬虫的全阶段需求，代码可直接部署至开发、生产环境，同时针对多品类数据特点定制拆分策略，兼顾性能、可维护性与业务适配性。

一、分库分表基础概念与项目整体规划

1.1 核心概念解析

1.1.1 垂直拆分（分库）

垂直拆分也称为按业务维度分库，依据数据所属业务品类、业务模块进行数据库拆分。在多品类爬虫场景中，将资讯、电商、招聘、舆情等不同品类的爬虫数据，分别存储到独立数据库中。拆分后每个数据库只承载单一或少数几个品类的数据，实现业务资源隔离，解决单库压力过大、数据混杂的问题。垂直拆分不会改变单表的数据量，核心作用是做业务解耦与库级负载分流。

1.1.2 水平拆分（分表）

水平拆分也称为按数据维度分表，在同一个数据库内部，按照指定规则将一张大表拆分为多张结构完全一致的数据表。拆分依据通常为数据 ID、采集时间、哈希值等，原始单表的海量数据被均匀分散到多张子表中，每张子表仅存储部分数据，以此降低单表数据体量，优化索引查询、数据写入效率。水平拆分不区分业务品类，核心作用是解决单表数据量过大的性能瓶颈。

1.1.3 混合拆分

结合垂直分库与水平分表的优势，先按照业务品类垂直拆分为多个独立数据库，再在每个数据库内部，对核心数据表进行水平分表。这是海量多品类爬虫项目最主流的架构方案，同时实现业务隔离与数据分流，适配亿级以上采集数据存储场景。

1.2 项目核心目标

1. 结合多品类爬虫业务特性，设计合理的垂直分库、水平分表拆分规则与路由算法；
2. 基于 Python 实现分库分表通用数据读写框架，支持数据自动路由、增删改查全操作；
3. 完成多品类模拟海量数据生成，验证分库分表架构的读写稳定性与负载分流效果；
4. 实现历史单库单表数据向分库分表架构的平滑迁移，保障数据完整性与业务连续性；
5. 封装数据库连接池，优化多库多表场景下的连接管理，避免连接资源耗尽；
6. 总结分库分表运维规范、踩坑要点与架构扩展方案，适配生产环境长期运行。

1.3 技术栈选型

结合爬虫高并发写入、批量查询、多库切换的业务特点，本项目技术组件及功能说明如下表：

表格

1.4 业务场景与拆分方案设计

本项目模拟四大主流爬虫品类：资讯爬虫、电商爬虫、招聘爬虫、舆情爬虫，基于业务特点制定分层拆分方案：

1. 第一层：垂直分库。为每一个爬虫品类单独创建专属数据库，即资讯库spider_news_db、电商库spider_shop_db、招聘库spider_job_db、舆情库spider_senti_db，实现不同品类数据物理隔离。
2. 第二层：水平分表。每个品类数据库内，将核心业务数据表按照主键 ID 取模规则拆分为 8 张子表，表结构完全一致，分散单表写入压力。
3. 路由规则。新增数据时，根据数据唯一 ID 计算路由，自动判定数据所属数据库与数据表；查询数据时，依据查询条件反向定位目标库、目标表，完成精准读写。

二、环境搭建与数据库基础准备

2.1 依赖库安装

执行 pip 命令安装项目所需第三方库，国内环境推荐使用清华镜像源加速下载：

bash

运行

# 常规安装 pip install pymysql dbutils faker # 镜像源加速安装 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pymysql dbutils faker

安装完成后，在 Python 交互环境执行导入语句，无报错即为环境配置正常：

python

运行

import pymysql from dbutils.pooled_db import PooledDB from faker import Faker

2.2 多数据库与分表结构创建

按照垂直分库规划，在 MySQL 中创建四个品类专属数据库，同时在每个数据库内批量创建 8 张结构一致的水平分表。所有分表字段统一，贴合爬虫通用数据结构，包含自增 ID、品类标识、标题、内容、来源、采集时间等核心字段。

2.2.1 批量创建数据库 SQL

sql

-- 资讯爬虫数据库 CREATE DATABASE IF NOT EXISTS spider_news_db DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 电商爬虫数据库 CREATE DATABASE IF NOT EXISTS spider_shop_db DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 招聘爬虫数据库 CREATE DATABASE IF NOT EXISTS spider_job_db DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 舆情爬虫数据库 CREATE DATABASE IF NOT EXISTS spider_senti_db DEFAULT CHARSET utf8mb4 COLLATE utf8mb4_unicode_ci;

2.2.2 单库分表结构（以资讯库为例，其余库表结构完全一致）

每个数据库内创建spider_data_0至spider_data_7共 8 张分表，表结构统一，以下为建表语句：

sql

USE spider_news_db; -- 批量创建8张水平分表，表后缀从0到7 CREATE TABLE IF NOT EXISTS spider_data_0 ( id BIGINT NOT NULL COMMENT '数据唯一主键ID', category VARCHAR(30) NOT NULL COMMENT '数据品类', title VARCHAR(500) NOT NULL COMMENT '数据标题', content TEXT COMMENT '采集正文内容', source VARCHAR(100) COMMENT '数据来源站点', crawl_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '爬虫采集时间', INDEX idx_id (id), INDEX idx_crawl_time (crawl_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '爬虫分表0'; CREATE TABLE IF NOT EXISTS spider_data_1 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_2 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_3 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_4 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_5 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_6 LIKE spider_data_0; CREATE TABLE IF NOT EXISTS spider_data_7 LIKE spider_data_0;

其余spider_shop_db、spider_job_db、spider_senti_db三个数据库，执行完全相同的分表创建语句，保证全部分表结构统一。

2.3 数据库连接池配置原理

分库分表架构下，程序需要频繁切换多个数据库连接，频繁创建、销毁连接会带来大量性能开销。DBUtils 连接池会预先创建一定数量的数据库连接并缓存，程序取用连接完成操作后归还至池子，实现连接复用。同时可配置最大连接数、空闲超时时间，避免连接溢出，是多库场景的必备组件。

三、分库分表路由算法设计与实现

路由算法是分库分表的核心，决定数据存储到哪一个数据库、哪一张数据表。本项目采用品类定向分库 + ID 取模分表的组合路由规则，逻辑简单、性能高效，适配爬虫海量写入场景。

3.1 路由规则定义

1. 分库规则：通过数据所属category品类字段直接映射目标数据库名称，固定映射关系如下：
   • news → spider_news_db（资讯库）
   • shop → spider_shop_db（电商库）
   • job → spider_job_db（招聘库）
   • senti → spider_senti_db（舆情库）
2. 分表规则：获取数据唯一主键id，对分表总数8进行取模运算，计算结果即为分表后缀编号。计算公式：table_suffix = id % 8。
3. 最终路由结果：数据库名称 + 固定表名前缀 + 计算得到的表后缀。

3.2 路由工具类代码实现

封装独立路由工具函数，解耦路由逻辑与业务读写逻辑，便于后续规则修改与维护。

python

运行

# 分库映射字典：品类 -> 数据库名 DB_MAPPING = { "news": "spider_news_db", "shop": "spider_shop_db", "job": "spider_job_db", "senti": "spider_senti_db" } # 分表总数 TABLE_COUNT = 8 # 分表名称前缀 TABLE_PREFIX = "spider_data_" def get_target_db(category: str) -> str: """ 根据品类获取目标数据库名称 :param category: 数据品类标识 :return: 数据库名称 """ return DB_MAPPING.get(category, "") def get_target_table(data_id: int) -> str: """ 根据数据ID获取目标分表名称 :param data_id: 数据唯一主键ID :return: 完整数据表名 """ suffix = data_id % TABLE_COUNT return f"{TABLE_PREFIX}{suffix}" def get_full_route(category: str, data_id: int) -> tuple: """ 一次性获取完整路由信息：数据库名、数据表名 :return: (db_name, table_name) """ db_name = get_target_db(category) table_name = get_target_table(data_id) return db_name, table_name

3.3 路由算法原理解析

1. 品类分库原理：采用静态映射方式，无复杂计算，查询与路由速度最快。依托业务品类做垂直拆分，天然契合多品类爬虫的业务边界，运维人员可直接按数据库区分业务数据。
2. ID 取模分表原理：取模运算属于常数级时间复杂度，计算效率极高。当 ID 自增且分布均匀时，数据会近似均匀分散到 8 张子表中，有效均衡每张分表的数据量与读写压力。该算法适合 ID 连续自增的爬虫主键场景。
3. 算法局限性说明：若后续需要增加分表数量，原有数据路由会全部失效，因此前期需根据预估数据总量合理规划分表数量。若业务存在分表扩容需求，可后续切换为一致性哈希路由算法。

四、数据库连接池封装

基于 DBUtils 实现全局数据库连接池，统一管理所有数据库连接，支持动态切换目标数据库，为后续数据读写提供连接支撑。

4.1 连接池完整代码

python

运行

from dbutils.pooled_db import PooledDB import pymysql # 数据库基础配置 MYSQL_HOST = "127.0.0.1" MYSQL_USER = "root" MYSQL_PWD = "你的数据库密码" MYSQL_PORT = 3306 MYSQL_CHARSET = "utf8mb4" # 连接池全局配置 POOL_MIN_CACHED = 2 # 最小空闲连接数 POOL_MAX_CACHED = 10 # 最大空闲连接数 POOL_MAX_USAGE = 0 # 单个连接最大复用次数，0表示无限制 # 初始化全局连接池 pool = PooledDB( creator=pymysql, mincached=POOL_MIN_CACHED, maxcached=POOL_MAX_CACHED, maxusage=POOL_MAX_USAGE, host=MYSQL_HOST, user=MYSQL_USER, password=MYSQL_PWD, port=MYSQL_PORT, charset=MYSQL_CHARSET ) def get_db_connection(db_name: str): """ 从连接池获取指定数据库的连接 :param db_name: 目标数据库名称 :return: 数据库连接对象、游标对象 """ if not db_name: raise ValueError("数据库名称不能为空") conn = pool.connection() # 切换至目标数据库 conn.select_db(db_name) cursor = conn.cursor(pymysql.cursors.DictCursor) return conn, cursor def close_connection(conn, cursor): """归还连接至连接池，关闭游标""" cursor.close() conn.close()

4.2 连接池核心原理

1. 程序启动时，连接池预先创建最小数量的空闲连接，等待业务调用；业务请求到来时，直接取出空闲连接，省去 TCP 握手、权限校验等创建开销。
2. select_db()方法实现在同一连接上动态切换数据库，无需重复创建新连接，适配分库场景。
3. 连接使用完毕后调用close()并非真正销毁连接，而是将连接放回连接池复用，只有程序终止或连接超时才会彻底释放。

五、分库分表数据增删改查实现

结合路由工具类与连接池，封装完整的数据写入、单条查询、批量查询、数据更新、数据删除五大核心操作，所有操作自动完成路由，开发者无需手动指定库表。

5.1 模拟爬虫数据生成工具

使用 Faker 生成多品类模拟爬虫数据，用于功能测试与压力验证：

python

运行

from faker import Faker import random fake = Faker(locale="zh_CN") # 品类列表 CATEGORY_LIST = ["news", "shop", "job", "senti"] def generate_spider_data(data_id: int) -> dict: """生成单条模拟爬虫数据""" category = random.choice(CATEGORY_LIST) data = { "id": data_id, "category": category, "title": fake.sentence(nb_words=8), "content": fake.paragraph(nb_sentences=3), "source": fake.domain_name() } return data

5.2 数据插入（爬虫核心写入场景）

爬虫采集数据后自动路由至对应库表，支持单条插入与批量插入，批量插入大幅提升写入效率。

python

运行

def insert_single_data(data: dict): """单条数据插入分库分表""" data_id = data["id"] category = data["category"] # 路由计算 db_name, table_name = get_full_route(category, data_id) conn, cursor = get_db_connection(db_name) try: sql = f""" INSERT INTO {table_name} (id, category, title, content, source) VALUES (%s, %s, %s, %s, %s) """ params = (data["id"], data["category"], data["title"], data["content"], data["source"]) cursor.execute(sql, params) conn.commit() print(f"数据插入成功，路由：{db_name}.{table_name}，ID：{data_id}") except Exception as e: conn.rollback() print(f"数据插入失败：{str(e)}") finally: close_connection(conn, cursor) def batch_insert_data(start_id: int, count: int): """批量生成并插入海量测试数据""" for i in range(count): current_id = start_id + i data = generate_spider_data(current_id) insert_single_data(data)

原理说明：写入流程遵循「生成数据→计算路由→获取连接→执行 SQL→提交事务→归还连接」，每一条数据都会根据自身 ID 与品类精准落入对应分库分表。事务机制保证单条数据写入的原子性，失败则自动回滚。

5.3 单条数据查询（根据 ID 精准查询）

已知数据 ID 与品类，反向路由定位库表，实现精准查询，是爬虫数据回溯、详情查看的常用场景。

python

运行

def query_single_data(category: str, data_id: int): """根据品类和ID查询单条数据""" db_name, table_name = get_full_route(category, data_id) conn, cursor = get_db_connection(db_name) try: sql = f"SELECT * FROM {table_name} WHERE id = %s" cursor.execute(sql, (data_id,)) result = cursor.fetchone() return result finally: close_connection(conn, cursor)

5.4 数据更新与删除

基于路由规则定位目标库表，执行更新、删除操作，语法与单表操作一致。

python

运行

def update_data(category: str, data_id: int, new_title: str): """更新数据标题字段""" db_name, table_name = get_full_route(category, data_id) conn, cursor = get_db_connection(db_name) try: sql = f"UPDATE {table_name} SET title = %s WHERE id = %s" cursor.execute(sql, (new_title, data_id)) conn.commit() print("数据更新成功") except Exception as e: conn.rollback() print(f"数据更新失败：{str(e)}") finally: close_connection(conn, cursor) def delete_data(category: str, data_id: int): """删除指定ID数据""" db_name, table_name = get_full_route(category, data_id) conn, cursor = get_db_connection(db_name) try: sql = f"DELETE FROM {table_name} WHERE id = %s" cursor.execute(sql, (data_id,)) conn.commit() print("数据删除成功") except Exception as e: conn.rollback() print(f"数据删除失败：{str(e)}") finally: close_connection(conn, cursor)

5.5 全表遍历查询（批量统计场景）

当需要统计某一品类全量数据时，需要遍历该品类数据库下所有 8 张分表，汇总查询结果。

python

运行

def query_all_table_data(category: str): """遍历品类下所有分表，汇总全部数据""" db_name = get_target_db(category) all_result = [] conn, cursor = get_db_connection(db_name) try: for suffix in range(TABLE_COUNT): table_name = f"{TABLE_PREFIX}{suffix}" sql = f"SELECT * FROM {table_name}" cursor.execute(sql) res = cursor.fetchall() all_result.extend(res) return all_result finally: close_connection(conn, cursor)

原理说明：无 ID 条件的全量查询无法精准路由，必须遍历当前品类下所有分表，再在程序层合并结果。该方式适合后台统计、数据导出等低频场景，高并发业务查询需尽量使用 ID 精准路由。

六、单库单表数据迁移至分库分表

存量业务升级场景中，需要将历史单库单表的海量爬虫数据，平滑迁移至新的分库分表架构，本节实现完整的数据迁移脚本，保证数据零丢失、业务无中断。

6.1 迁移思路

1. 建立临时迁移数据库，存放历史原始数据；
2. 分页读取原始大表数据，避免一次性加载全量数据导致内存溢出；
3. 逐条计算路由，将数据写入新的分库分表架构；
4. 迁移完成后对比新旧数据总量、关键字段，校验数据完整性。

6.2 数据迁移代码实现

python

运行

def migrate_old_data(old_db: str, old_table: str, page_size: int = 1000): """ 旧单表数据迁移至分库分表 :param old_db: 原始数据库名 :param old_table: 原始数据表名 :param page_size: 分页读取条数 """ conn, cursor = get_db_connection(old_db) offset = 0 total_migrate = 0 try: while True: # 分页读取原始数据 sql = f"SELECT id,category,title,content,source FROM {old_table} LIMIT %s,%s" cursor.execute(sql, (offset, page_size)) rows = cursor.fetchall() if not rows: break # 逐条写入分库分表 for row in rows: data = dict(row) insert_single_data(data) total_migrate += 1 offset += page_size print(f"数据迁移完成，总迁移条数：{total_migrate}") finally: close_connection(conn, cursor)

6.3 迁移注意事项

1. 迁移操作选择业务低峰期执行，避免影响线上爬虫正常写入；
2. 采用分页读取，严格控制单次加载数据量，防止内存溢出；
3. 迁移完成后执行数据条数比对、抽样校验，确认数据完整；
4. 迁移期间新旧架构双写，平稳切换业务流量，实现灰度上线。

七、架构性能测试与问题优化

7.1 压力测试

调用批量插入函数生成万级、十万级测试数据，观察数据库负载、写入耗时。分库分表架构下，单表数据量被控制在合理范围，索引查询、写入速度远优于传统单表架构。

python

运行

# 从ID=1开始，批量生成10000条测试数据 batch_insert_data(start_id=1, count=10000)

7.2 常见问题与优化方案

7.2.1 跨库跨表联表查询

分库分表后，不同数据库、不同分表之间无法直接使用 MySQL 联表查询。优化方案：业务层做数据关联、冗余字段设计、Elasticsearch 搜索引擎辅助查询。

7.2.2 自增 ID 冲突问题

多表写入时，原生数据库自增主键会出现 ID 重复。优化方案：使用雪花算法、UUID、数据库分段自增方案生成全局唯一 ID。

7.2.3 分表扩容问题

基于取模算法的分表，后期增加分表数量会导致路由错乱。优化方案：前期预留足够分表数量，或切换为一致性哈希路由算法。

7.2.4 事务失效问题

跨库操作无法使用 MySQL 本地事务。优化方案：核心业务引入分布式事务，非核心业务做最终一致性补偿。

7.3 索引优化规范

1. 每张分表必须为路由字段（id）、查询高频字段建立索引；
2. 避免在分表中创建过多冗余索引，降低写入开销；
3. 时间范围查询场景，优先使用采集时间字段分区表结合分表使用。

八、项目扩展与生产环境运维规范

8.1 架构扩展方向

1. 读写分离：为主库配置从库，主库负责爬虫数据写入，从库承担查询、统计、报表等读请求，进一步分流压力；
2. 中间件整合：引入 Sharding-JDBC、MyCat 等专业分库分表中间件，替代手动路由代码，降低开发复杂度；
3. 冷热数据分离：将近期活跃的热数据存储在高性能 MySQL，历史冷数据迁移至时序数据库、对象存储，节省存储成本；
4. 分区表结合：在水平分表基础上，按时间做表内分区，兼顾分表与分区的双重优势。

8.2 生产环境运维规范

1. 权限管控：不同品类数据库分配独立账号，最小权限原则管控访问权限；
2. 定时备份：沿用前文 MySQL 定时备份方案，对每一个分库单独配置备份策略，保障数据安全；
3. 监控告警：监控数据库 CPU、内存、磁盘、连接数、慢查询，异常时及时告警；
4. 容量规划：定期统计每张分表的数据增长量，提前预判存储瓶颈，及时扩容；
5. 版本迭代：表结构变更、字段新增时，所有分表同步执行 DDL 语句，保证结构统一。