当前位置：首页 > news >正文

Python爬虫数据存储：MySQL实战教程

news 2025/10/25 10:28:59

一、环境准备：搭建爬虫存储一体化环境

1. 数据库安装与配置

2. Python依赖安装

3. 数据库连接池配置

二、数据建模：设计爬虫专用表结构

1. 豆瓣电影数据表设计

2. 索引优化策略

三、爬虫开发：从抓取到存储的全流程

1. 豆瓣Top250爬取实现

2. 数据清洗与转换

3. 批量存储实现

四、进阶技巧：提升存储性能与可靠性

1. 事务处理最佳实践

2. 重复数据处理策略

3. 大数据量存储优化

五、完整案例：豆瓣电影数据采集系统

1. 系统架构设计

2. 核心代码实现

3. 监控与告警

六、常见问题Q&A

七、总结与建议

免费编程软件「python+pycharm」
链接：https://pan.quark.cn/s/48a86be2fdc0

在数据驱动的时代，爬虫技术已成为获取公开信息的重要手段。但抓取到的数据若不能有效存储，就如同收集了散落的珍珠却缺少串起它们的丝线。MySQL作为成熟的开源关系型数据库，凭借其稳定性、事务支持和社区生态，成为爬虫数据存储的首选方案。本文将以豆瓣电影Top250数据抓取为例，演示如何将爬取的结构化数据高效存入MySQL，并处理实际开发中遇到的常见问题。

一、环境准备：搭建爬虫存储一体化环境

1. 数据库安装与配置

推荐使用Docker快速部署MySQL 8.0：

docker run --name mysql_crawler -e MYSQL_ROOT_PASSWORD=123456 -p 3306:3306 -d mysql:8.0

连接数据库后执行基础优化：

CREATE DATABASE crawler_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
-- 提升批量插入性能
SET GLOBAL innodb_buffer_pool_size=1G;

utf8mb4字符集可完整存储emoji等特殊字符，避免中文乱码问题。

2. Python依赖安装

pip install pymysql requests beautifulsoup4 sqlalchemy

pymysql：纯Python实现的MySQL驱动
sqlalchemy：ORM工具，简化数据库操作
beautifulsoup4：HTML解析库

3. 数据库连接池配置

使用SQLAlchemy创建连接池：

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:123456@localhost/crawler_db',pool_size=5,       # 基础连接数max_overflow=10,   # 最大溢出连接数pool_recycle=3600  # 连接回收时间(秒)
)

某电商爬虫系统实测显示，合理配置连接池后，数据库操作吞吐量提升3倍。

二、数据建模：设计爬虫专用表结构

1. 豆瓣电影数据表设计

分析目标数据后创建三张表：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, Text, ForeignKeyBase = declarative_base()class Movie(Base):__tablename__ = 'movies'id = Column(Integer, primary_key=True)title = Column(String(100), nullable=False)rating = Column(Float)quote = Column(String(200))info = Column(Text)  # 存储导演/主演/年份等拼接信息class Comment(Base):__tablename__ = 'comments'id = Column(Integer, primary_key=True)movie_id = Column(Integer, ForeignKey('movies.id'))user = Column(String(50))content = Column(Text)stars = Column(Integer)  # 1-5星评分class Genre(Base):__tablename__ = 'genres'id = Column(Integer, primary_key=True)name = Column(String(30), unique=True)

通过外键关联建立数据关系，便于后续分析电影类型分布。

2. 索引优化策略

为高频查询字段建立索引：

class Movie(Base):__tablename__ = 'movies'# ...其他字段...__table_args__ = (Index('idx_rating', 'rating'),  # 评分排序查询Index('idx_title', 'title')     # 标题搜索)

实测显示，为rating字段添加索引后，按评分筛选电影的速度提升5倍。

三、爬虫开发：从抓取到存储的全流程

1. 豆瓣Top250爬取实现

import requests
from bs4 import BeautifulSoupdef fetch_movie_list(page):url = f'https://movie.douban.com/top250?start={(page-1)*25}'headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}resp = requests.get(url, headers=headers)soup = BeautifulSoup(resp.text, 'html.parser')movies = []for item in soup.select('.item'):title = item.select_one('.title').textrating = float(item.select_one('.rating_num').text)quote = item.select_one('.inq').text if item.select_one('.inq') else ''info = item.select_one('.bd').text.strip().replace('\n', ' ').replace(' ', ' ', 2)movies.append({'title': title,'rating': rating,'quote': quote,'info': info})return movies

关键点：

模拟浏览器请求头避免被封
使用CSS选择器精准定位元素
处理可能缺失的字段（如quote）

2. 数据清洗与转换

def parse_movie_info(info):# 示例输入："1994 / 美国 / 犯罪 剧情"parts = info.split('/')return {'year': parts[0].strip(),'country': parts[1].strip(),'genres': [g.strip() for g in parts[2].split()]}

将拼接的字符串信息拆分为结构化数据，便于后续分析。

3. 批量存储实现

使用ORM进行高效存储：

from sqlalchemy.orm import sessionmakerSession = sessionmaker(bind=engine)
session = Session()def store_movies(movies):try:# 批量插入电影信息for movie_data in movies:movie = Movie(title=movie_data['title'],rating=movie_data['rating'],quote=movie_data['quote'],info=movie_data['info'])session.add(movie)# 分批提交（每50条提交一次）if len(movies) >= 50:session.flush()session.commit()except Exception as e:session.rollback()print(f"存储失败: {e}")finally:session.close()

批量提交策略可显著提升存储效率，某新闻爬虫系统测试表明，每100条提交一次比单条提交快8倍。

四、进阶技巧：提升存储性能与可靠性

1. 事务处理最佳实践

对于关联操作使用事务：

def add_movie_with_genres(movie_data, genres):session = Session()try:# 插入电影movie = Movie(title=movie_data['title'], ...)session.add(movie)session.flush()  # 获取自增ID# 插入类型关联（假设有中间表movie_genres）for genre_name in genres:genre = session.query(Genre).filter_by(name=genre_name).first()if not genre:genre = Genre(name=genre_name)session.add(genre)# 建立关联...session.commit()except:session.rollback()raise

事务确保电影和类型数据要么全部成功，要么全部回滚。

2. 重复数据处理策略

使用INSERT IGNORE或ON DUPLICATE KEY UPDATE：

# 方法1：ORM方式（需设置唯一约束）
from sqlalchemy.dialects.mysql import insertstmt = insert(Movie).values(title='肖申克的救赎',rating=9.7
)
stmt = stmt.on_duplicate_key_update(rating=9.7)  # 存在则更新
engine.execute(stmt)

适用于需要更新已有数据的场景，如价格监控爬虫。

3. 大数据量存储优化

对于百万级数据，采用分表策略：

# 按年份分表示例
def get_table_name(year):return f'movies_{year}'# 动态建表
if not engine.dialect.has_table(engine, 'movies_2023'):Movie2023 = type('Movie2023', (Base,), {'__tablename__': 'movies_2023','__table_args__': {'extend_existing': True},# ...表结构...})Base.metadata.create_all(engine, tables=[Movie2023.__table__])

某金融数据平台通过年度分表，将单表查询速度从12秒提升至0.8秒。

五、完整案例：豆瓣电影数据采集系统

1. 系统架构设计

爬虫节点 → 数据清洗 → MySQL主库 → 同步至分析库↓备用存储(SQLite)

主库处理写入，分析库负责查询
备用存储防止主库故障时数据丢失

2. 核心代码实现

def crawl_all_pages():all_movies = []for page in range(1, 11):  # Top250共10页movies = fetch_movie_list(page)all_movies.extend(movies)time.sleep(2)  # 礼貌爬取# 数据清洗processed = []for movie in all_movies:info = parse_movie_info(movie['info'])processed.append({**movie,'year': info['year'],'genres': info['genres']})# 存储到MySQLstore_movies(processed)# 同时存入SQLite备用sqlite_conn = sqlite3.connect('backup.db')# ...SQLite存储逻辑...

3. 监控与告警

添加简单的监控机制：

def check_storage_status():session = Session()movie_count = session.query(Movie).count()if movie_count < 250:send_alert(f"数据不完整，当前仅{movie_count}条")session.close()

六、常见问题Q&A

Q1：被网站封IP怎么办？
A：立即启用备用代理池，建议使用住宅代理（如站大爷IP代理），配合每请求更换IP策略。代码示例：

import random
PROXIES = [{'http': 'http://1.1.1.1:8080'},{'http': 'http://2.2.2.2:8080'}
]def get_random_proxy():return random.choice(PROXIES)resp = requests.get(url, proxies=get_random_proxy())

Q2：如何处理反爬机制？
A：综合使用以下策略：