爬虫实战PyCharm+Scrapy爬取数据并存入MySQL
一、背景与问题
在数据驱动的现代应用中,数据采集是构建数据仓库和分析系统的重要环节。传统爬虫方案常面临两大挑战:数据解析效率不足和数据存储结构不统一。Scrapy作为Python领域最成熟的爬虫框架,其异步架构和内置的Item Pipeline机制,为解决这些问题提供了系统性方案。
以某电商平台商品信息爬取为例,传统方案可能需要手动处理HTML解析、数据清洗、数据库连接等环节,导致代码冗余且维护困难。本文将深入解析Scrapy与MySQL的集成方案,通过实际案例展示如何构建高可用的爬虫系统。
二、基本原理
1. Scrapy框架架构
Scrapy采用典型的生产者-消费者模型,其核心组件包括:
- Spider:负责发送HTTP请求和解析响应内容
- Engine:协调各组件的工作流程
- Downloader:处理HTTP请求和响应
- Item Pipeline:负责数据清洗、验证和存储
- Middleware:处理请求和响应的中间件
其核心流程如下:
- Spider生成初始请求(Request)并发送给Engine
- Engine将请求分发给Downloader获取响应(Response)
- Engine将响应传递给Spider进行解析,提取Item或生成新的请求
- Engine将Item传递给Item Pipeline进行处理
2. MySQL存储机制
MySQL通过InnoDB引擎支持事务处理,其核心特征包括:
- ACID特性:保证数据一致性和完整性
- 索引优化:通过B+树结构加速查询
- 连接池机制:避免频繁创建/销毁连接
Scrapy与MySQL的集成需要处理三个关键问题:
- 爬虫并发与数据库连接的资源竞争
- 数据类型转换与字段校验
- 批量插入的性能优化
三、环境准备
1. 软件环境
# 安装Scrapy和MySQL驱动
pip install scrapy pymysql
# 创建MySQL数据库
CREATE DATABASE scraping_db;
USE scraping_db;
# 创建数据表(以商品信息为例)
CREATE TABLE products (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
price DECIMAL(10,2),
category VARCHAR(100),
stock INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
2. 项目结构
scrapy_project/
├── scrapy.cfg
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py
├── spiders/
│ └── book_spider.py
└── db_utils.py
四、核心实现
1. 爬虫逻辑实现(book_spider.py)
import scrapy
class BookSpider(scrapy.Spider):
name = 'book_spider'
start_urls = ['https://example-books.com/page/1', 'https://example-books.com/page/2']
def parse(self, response):
for book in response.css('div.book'):
yield {
'name': book.css('h2::text').get(),
'price': float(book.css('span.price::text').get()),
'category': book.css('span.category::text').get(),
'stock': int(book.css('span.stock::text').get())
}
关键点说明:
- 使用CSS选择器高效解析HTML
- 自动类型转换(字符串转float/int)
- 返回的字典结构与Item Pipeline兼容
2. 数据管道实现(pipelines.py)
import pymysql
from scrapy.exceptions import DropItem
class MySQLPipeline:
def __init__(self, host, database, user, password, table):
self.host = host
self.database = database
self.user = user
self.password = password
self.table = table
self.connection = None
@classmethod
def from_crawler(cls, crawler):
return cls(
host=crawler.settings.get('MYSQL_HOST'),
database=crawler.settings.get('MYSQL_DATABASE'),
user=crawler.settings.get('MYSQL_USER'),
password=crawler.settings.get('MYSQL_PASSWORD'),
table=crawler.settings.get('MYSQL_TABLE')
)
def open_spider(self, spider):
self.connection = pymysql.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database,
cursorclass=pymysql.cursors.DictCursor
)
self.create_table()
def close_spider(self, spider):
self.connection.close()
def create_table(self):
with self.connection.cursor() as cursor:
cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {self.table} (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
price DECIMAL(10,2),
category VARCHAR(100),
stock INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.connection.commit()
def process_item(self, item, spider):
try:
with self.connection.cursor() as cursor:
# 使用预编译语句防止SQL注入
sql = f"""
INSERT INTO {self.table}
(name, price, category, stock)
VALUES (%s, %s, %s, %s)
"""
cursor.execute(sql, (
item['name'],
item['price'],
item['category'],
item['stock']
))
self.connection.commit()
return item
except pymysql.MySQLError as e:
self.connection.rollback()
raise DropItem(f"插入数据库失败: {e}")
关键点说明:
- 使用连接池机制管理数据库连接
- 预编译语句防止SQL注入
- 事务回滚机制保证数据一致性
- 自动创建表结构(首次运行时)
3. 配置文件(settings.py)
# MySQL配置
MYSQL_HOST = 'localhost'
MYSQL_USER = 'scraping_user'
MYSQL_PASSWORD = 'securepassword'
MYSQL_DATABASE = 'scraping_db'
MYSQL_TABLE = 'products'
# 启用管道
ITEM_PIPELINES = {
'scrapy_project.pipelines.MySQLPipeline': 300
}
五、完整案例:图书信息爬取系统
1. 项目结构
scrapy_book_project/
├── scrapy.cfg
├── items.py
├── middlewares.py
├── pipelines.py
├── settings.py
├── spiders/
│ └── book_spider.py
└── db_utils.py
2. 爬虫逻辑优化
import scrapy
from scrapy.http import Request
from datetime import datetime
class BookSpider(scrapy.Spider):
name = 'book_spider'
start_urls = ['https://example-books.com/page/1', 'https://example-books.com/page/2']
custom_settings = {
'LOG_LEVEL': 'INFO'
}
def parse(self, response):
for book in response.css('div.book'):
yield {
'name': book.css('h2::text').get(),
'price': float(book.css('span.price::text').get()),
'category': book.css('span.category::text').get(),
'stock': int(book.css('span.stock::text').get()),
'timestamp': datetime.now().isoformat()
}
# 分页处理
next_page = response.css('a.next-page::attr(href)').get()
if next_page and 'page' in next_page:
yield Request(url=next_page, callback=self.parse)
3. 性能优化方案
# 优化后的MySQLPipeline
class MySQLPipeline:
def __init__(self, ...):
self.batch_size = 100 # 批量插入大小
self.buffer = []
def process_item(self, item, spider):
self.buffer.append(item)
if len(self.buffer) >= self.batch_size:
self.insert_batch()
return item
def insert_batch(self):
with self.connection.cursor() as cursor:
sql = f"""
INSERT INTO {self.table}
(name, price, category, stock, created_at)
VALUES (%s, %s, %s, %s, %s)
"""
cursor.executemany(sql, [
(
item['name'],
item['price'],
item['category'],
item['stock'],
item['timestamp']
)
for item in self.buffer
])
self.connection.commit()
self.buffer.clear()
六、源码解析
1. Scrapy的Request调度机制
# 在Spider中生成Request
yield Request(url, callback=self.parse)
- Scrapy使用优先队列管理请求
- 可通过
meta参数传递额外信息 - 支持自定义
dont_filter参数控制去重
2. MySQL连接池实现
# 在MySQLPipeline中使用连接池
self.connection = pymysql.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database,
cursorclass=pymysql.cursors.DictCursor,
connect_timeout=5
)
- 设置连接超时时间防止阻塞
- 使用
pymysql库的连接池特性 - 在
close_spider中确保连接关闭
七、进阶使用
1. 分布式爬虫方案
# 在settings.py中配置
SPIDER_MIDDLEWARES = {
'scrapy.contrib.spidermiddleware.offsite.OffsiteMiddleware': 500,
'scrapy.contrib.spidermiddleware.referer.RefererMiddleware': 700
}
- 使用
scrapy_redis实现分布式爬虫 - 通过Redis队列管理请求
- 支持多节点部署
2. 异步处理优化
# 在settings.py中配置
DOWNLOAD_DELAY = 1
CONCURRENT_REQUESTS = 16
- 控制并发请求数量
- 设置请求间隔防止被封
- 可结合
scrapy-splash处理JavaScript渲染
八、性能与工程实践
1. 性能优化策略
| 优化措施 | 说明 | 效果 |
|---|
| 批量插入 | 减少数据库事务次数 | 提升3-5倍写入速度 |
| 缓存中间结果 | 减少重复计算 | 降低CPU占用 |
| 优化CSS选择器 | 减少DOM遍历 | 提升解析速度 |
| 使用连接池 | 避免频繁连接 | 降低延迟 |
2. 异常处理机制
# 在process_item中添加异常处理
def process_item(self, item, spider):
try:
# 爬虫逻辑
except Exception as e:
self.logger.error(f"处理Item失败: {e}")
return item # 返回Item继续处理
3. 安全防护措施
- 使用
scrapy-splash处理动态内容 - 设置
USER_AGENT防止被识别 - 使用
scrapy-captcha处理验证码 - 配置
HTTP_PROXY进行流量控制
九、常见问题与踩坑
1. 常见错误及解决办法
| 错误类型 | 错误信息 | 解决方案 |
|---|
| 连接失败 | "Access denied for user" | 检查MySQL用户权限 |
| 写入失败 | "Duplicate entry" | 添加唯一索引并处理冲突 |
| 性能低下 | "Timeout expired" | 增加连接超时时间 |
| 数据丢失 | "Aborted connection" | 增加重试机制 |
2. 典型问题分析
问题:爬虫频繁被封禁
# 原始配置
USER_AGENT = 'Mozilla/5.0'
DOWNLOAD_DELAY = 0
改进方案:
# 增加随机延迟
DOWNLOAD_DELAY = 2
RANDOMIZE_DOWNLOAD_DELAY = True
问题:数据类型不匹配
# 错误代码
cursor.execute("INSERT INTO products (price) VALUES (%s)", (item['price'],))
改进方案:
# 显式指定字段类型
cursor.execute("INSERT INTO products (price) VALUES (%s)", (float(item['price']),))
十、最佳实践
1. 架构设计建议
- 分层设计:Spider负责解析,Pipeline负责存储
- 模块化开发:将不同功能拆分为独立组件
- 配置分离:敏感信息通过环境变量管理
2. 性能调优建议
- 使用
pymysql的connect()参数配置连接池 - 在Pipeline中启用批量插入
- 使用
scrapy-redis实现分布式爬虫
3. 安全最佳实践
- 对所有输入进行校验
- 使用
scrapy-splash处理JavaScript - 设置合理的请求频率
- 对敏感信息进行加密存储
十一、总结
Scrapy与MySQL的集成方案,通过其异步架构和内置的Item Pipeline机制,为数据采集提供了高效的解决方案。在实际应用中,我们应根据需求选择合适的实现方式:
- 适用场景:大规模数据采集、结构化数据存储、需要高并发处理的场景
- 不适用场景:需要处理动态内容、频繁更新的数据、对响应时间要求极高的场景
通过合理配置、性能优化和安全防护,可以构建稳定可靠的爬虫系统。建议在实际项目中结合具体需求,灵活运用本文介绍的方案,同时注意遵守相关法律法规和网站的robots.txt规则。