Scrapy 实战：编写一个分布式爬虫爬取海量数据

小狐狸S

1586人浏览 · 2025-10-05 11:36:08

小狐狸S · 2025-10-05 11:36:08 发布

在数据驱动的时代，获取海量数据成为企业和研究者的核心需求。当单一爬虫无法应对大规模数据采集任务时，分布式爬虫应运而生。Scrapy 作为 Python 生态中最强大的爬虫框架之一，结合分布式技术可以轻松实现海量数据的高效爬取。本文将带你从零开始，构建一个基于 Scrapy 的分布式爬虫系统。

一、分布式爬虫的核心原理

分布式爬虫通过多台机器或多个进程协同工作，共同完成数据采集任务，其核心优势在于：

突破单机性能瓶颈，大幅提升爬取效率
分散爬虫压力，降低被目标网站反爬的概率
具备容错能力，单节点故障不影响整体任务

Scrapy 本身并不直接支持分布式，但通过结合Scrapy-Redis扩展，可以实现基于 Redis 的分布式调度，让多台机器共享爬取队列和去重集合。

二、环境准备

1. 基础依赖安装

bash

# 安装Scrapy
pip install scrapy

# 安装Scrapy-Redis扩展
pip install scrapy-redis

# 安装Redis服务器（用于分布式协调）
# Ubuntu/Debian
sudo apt-get install redis-server
# CentOS
sudo yum install redis
# macOS
brew install redis

2. 环境配置

启动 Redis 服务并确保所有爬虫节点可访问（修改redis.conf中的bind配置）
确保所有节点的 Python 环境一致，避免依赖冲突
准备至少 2 台服务器（或本地多进程模拟）作为爬虫节点

三、分布式爬虫实现步骤

1. 创建 Scrapy 项目

bash

scrapy startproject distributed_crawler
cd distributed_crawler

2. 修改项目配置（settings.py）

核心是集成 Scrapy-Redis 的分布式组件：

python

运行

# 启用Redis调度存储，替代默认的内存调度
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 启用Redis去重，替代默认的RFPDupeFilter
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# Redis连接配置（替换为你的Redis服务器地址）
REDIS_URL = "redis://192.168.1.100:6379/0"

# 允许暂停和恢复爬取
SCHEDULER_PERSIST = True

# 爬取队列类型（优先级队列）
SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.PriorityQueue"

# 并发设置（根据服务器性能调整）
CONCURRENT_REQUESTS = 100
DOWNLOAD_DELAY = 0.5

# 启用cookie保持（视目标网站而定）
COOKIES_ENABLED = False

# 下载器中间件（添加随机User-Agent等反爬措施）
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
    'scrapy_fake_useragent.middleware.RandomUserAgentMiddleware': 400,
}

# 项目管道（数据存储）
ITEM_PIPELINES = {
    # 可选：将数据暂存到Redis
    'scrapy_redis.pipelines.RedisPipeline': 300,
    # 自定义管道：存储到数据库
    'distributed_crawler.pipelines.MysqlPipeline': 400,
}

3. 编写爬虫代码（spiders / 示例爬虫.py）

以爬取某电商商品数据为例：

python

运行

import scrapy
from scrapy_redis.spiders import RedisSpider
from distributed_crawler.items import ProductItem

class ProductSpider(RedisSpider):
    """分布式爬虫需继承RedisSpider而非普通Spider"""
    name = "product_spider"
    # 从Redis中获取起始URL（通过lpush命令添加）
    redis_key = "product:start_urls"

    def parse(self, response):
        # 解析列表页，提取商品详情页URL
        detail_urls = response.xpath('//a[@class="product-link"]/@href').getall()
        for url in detail_urls:
            yield scrapy.Request(url, callback=self.parse_detail)

        # 提取下一页URL
        next_page = response.xpath('//a[@class="next-page"]/@href').get()
        if next_page:
            yield scrapy.Request(next_page, callback=self.parse)

    def parse_detail(self, response):
        # 解析商品详情
        item = ProductItem()
        item['id'] = response.xpath('//div[@id="product-id"]/text()').get()
        item['name'] = response.xpath('//h1[@class="product-name"]/text()').get()
        item['price'] = response.xpath('//span[@class="price"]/text()').get()
        item['sales'] = response.xpath('//span[@class="sales-count"]/text()').get()
        yield item

4. 定义数据模型（items.py）

python

运行

import scrapy

class ProductItem(scrapy.Item):
    id = scrapy.Field()
    name = scrapy.Field()
    price = scrapy.Field()
    sales = scrapy.Field()

5. 实现数据存储管道（pipelines.py）

python

运行

import pymysql

class MysqlPipeline:
    def __init__(self):
        # 数据库连接配置
        self.db_params = {
            'host': '192.168.1.101',
            'user': 'crawler',
            'password': 'password',
            'database': 'product_data',
            'port': 3306
        }
        self.conn = None
        self.cursor = None

    def open_spider(self, spider):
        self.conn = pymysql.connect(**self.db_params)
        self.cursor = self.conn.cursor()

    def process_item(self, item, spider):
        # 插入数据
        sql = """
        INSERT INTO products (id, name, price, sales)
        VALUES (%s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE name=%s, price=%s, sales=%s
        """
        try:
            self.cursor.execute(sql, (
                item['id'], item['name'], item['price'], item['sales'],
                item['name'], item['price'], item['sales']
            ))
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            spider.logger.error(f"数据库错误: {e}")
        return item

    def close_spider(self, spider):
        self.cursor.close()
        self.conn.close()

四、启动分布式爬虫

1.** 启动 Redis 服务器 **```bashredis-server /etc/redis/redis.conf

plaintext


2.** 添加起始URL到Redis队列 **```bash
# 连接Redis
redis-cli -h 192.168.1.100

# 添加起始URL（根据实际目标网站修改）
lpush product:start_urls https://example.com/products?page=1

3.** 在多个节点启动爬虫 ** 在每台服务器上执行：

bash

cd distributed_crawler
scrapy crawl product_spider

此时，所有节点会自动从 Redis 获取任务并协同爬取，爬取结果将统一存储到 MySQL 数据库。

五、分布式爬虫优化策略

1.** 反爬机制应对 **- 配置随机 User-Agent（scrapy-fake-useragent）

使用代理 IP 池（可集成scrapy-proxies）
动态调整爬取频率（DOWNLOAD_DELAY）

2.** 性能优化 **- 根据服务器性能调整CONCURRENT_REQUESTS

启用 GZIP 压缩（HTTP_ACCEPT_ENCODING = 'gzip, deflate'）
使用异步数据库驱动（如aiomysql替代pymysql）

3.** 容错与监控 **- 配置爬虫自动重启（结合supervisor）

定期备份 Redis 数据（防止任务丢失）
集成监控工具（如 Prometheus+Grafana）监控爬取状态

六、注意事项

1.** 遵守网站 robots 协议 ，避免对目标服务器造成过大压力2. 分布式环境下确保所有节点时间同步 ，避免 Redis 数据冲突3. 大规模爬取前先进行小范围测试 ，验证爬虫逻辑和反爬策略4. 敏感数据爬取需遵守相关法律法规 **，避免法律风险

总结

基于 Scrapy+Redis 的分布式爬虫架构，通过共享任务队列和去重集合，实现了多节点协同工作，能高效应对海量数据爬取需求。在实际应用中，需根据目标网站特性不断优化爬虫策略，平衡爬取效率与反爬风险。随着数据规模增长，还可进一步扩展为弹性分布式架构，通过 Kubernetes 等容器编排工具实现爬虫节点的动态扩缩容，满足不同场景下的数据采集需求。