代码收藏家技术教程 2025-01-29

Python 网络爬虫高级教程：分布式爬取与大规模数据处理

经过基础爬虫和进阶爬虫的学习，我们已经掌握了爬虫的基本原理、动态内容处理及反爬机制的应对。然而，当我们面对海量数据或需要高效爬取多个站点时，分布式爬虫和数据存储、处理能力就显得尤为重要。本篇博客将带你迈向网络爬虫的高级阶段，学习分布式爬取、大规模数据处理以及性能优化。

一、分布式爬虫的概念

1. 什么是分布式爬虫？

分布式爬虫是指通过多个节点协作完成大规模网页爬取任务的爬虫架构。它能有效解决以下问题：

单台机器性能瓶颈（CPU、内存、网络带宽）。

爬取任务量大时的效率问题。

减少单个 IP 被封禁的风险。

2. 分布式爬虫的架构

分布式爬虫通常由以下部分组成：

任务队列：存储待爬取的 URL 列表。
多个爬虫节点：从任务队列中取出任务并执行爬取。
数据存储：将爬取的数据存储到数据库或文件中。
任务调度器：控制任务的分发和爬取状态。

二、使用 Scrapy 实现分布式爬虫

1. 安装 Scrapy 和 Scrapy-Redis

Scrapy 是一个强大的 Python 爬虫框架，支持分布式爬取。我们还需要 Scrapy-Redis 来实现任务队列的共享。

安装依赖库：

pip install scrapy scrapy-redis

2. Scrapy 项目结构

使用 Scrapy 创建一个新项目：

scrapy startproject distributed_crawler

项目结构：

distributed_crawler/
├── distributed_crawler/
│   ├── spiders/            # 爬虫文件目录
│   ├── __init__.py
│   ├── items.py            # 定义数据结构
│   ├── middlewares.py      # 爬虫中间件
│   ├── pipelines.py        # 数据存储管道
│   └── settings.py         # 项目配置
├── scrapy.cfg

3. 配置分布式功能

在 settings.py 中，配置 Scrapy-Redis：

# 使用 Scrapy-Redis 的调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 使用 Scrapy-Redis 的去重类
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# 配置 Redis 地址
REDIS_HOST = 'localhost'
REDIS_PORT = 6379

# 是否持久化队列，True 表示爬取完成后保留任务队列
SCHEDULER_PERSIST = True

4. 编写爬虫

在 spiders 目录下创建爬虫文件 quotes_spider.py：

import scrapy
from scrapy_redis.spiders import RedisSpider

class QuotesSpider(RedisSpider):
    name = "quotes"
    # 从 Redis 获取初始 URL
    redis_key = "quotes:start_urls"

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
            }

        # 爬取下一页
        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)

5. 配置 Redis 并运行

启动 Redis

确保 Redis 服务正在运行：

redis-server

添加起始 URL

将 URL 推送到 Redis：

redis-cli
> lpush quotes:start_urls http://quotes.toscrape.com

启动爬虫

运行爬虫：

scrapy crawl quotes

此时，多个爬虫节点可以通过共享 Redis 队列协作完成任务。

三、大规模数据处理与存储

分布式爬虫会产生大量数据，如何存储和处理这些数据是另一个重要问题。

1. 数据库存储

使用 MongoDB 存储数据

在 pipelines.py 中配置 MongoDB 数据存储：

import pymongo

class MongoPipeline:
    def open_spider(self, spider):
        self.client = pymongo.MongoClient("localhost", 27017)
        self.db = self.client["quotes_db"]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        self.db["quotes"].insert_one(dict(item))
        return item

在 settings.py 中启用管道：

ITEM_PIPELINES = {
    'distributed_crawler.pipelines.MongoPipeline': 300,
}

2. 数据清洗与分析

爬取数据后，可以使用 Pandas 进行数据清洗和分析：

import pandas as pd

# 读取数据
data = pd.read_json("quotes.json")

# 数据清洗
data = data.drop_duplicates()  # 去重

# 数据分析
author_counts = data["author"].value_counts()
print(author_counts)

3. 分布式存储与处理（可选）

对于更大规模的数据，可以使用分布式存储和处理工具：

存储：使用 Hadoop HDFS 或 Amazon S3。

处理：使用 Apache Spark 或 Hive。

四、性能优化

1. 增加爬虫节点

在分布式架构中，可以通过增加爬虫节点数量来提升效率。

配置多个节点时，只需共享相同的 Redis 配置，每个节点运行：

scrapy crawl quotes

2. 限制并发数

为了避免服务器封禁，可以限制爬取速度和并发数：

在 settings.py 中添加：

DOWNLOAD_DELAY = 1  # 请求间隔时间，单位秒
CONCURRENT_REQUESTS = 8  # 最大并发请求数

3. 错误重试与代理

错误重试

配置重试机制：

RETRY_ENABLED = True
RETRY_TIMES = 3  # 重试次数

使用代理池

通过免费或付费代理池切换 IP，防止被封禁。

DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 110,
    'distributed_crawler.middlewares.ProxyMiddleware': 100,
}

五、实战案例：分布式新闻爬虫

以下是一个分布式爬虫抓取新闻网站标题和链接的示例。

import scrapy
from scrapy_redis.spiders import RedisSpider

class NewsSpider(RedisSpider):
    name = "news"
    redis_key = "news:start_urls"

    def parse(self, response):
        for article in response.css("div.article"):
            yield {
                "title": article.css("h2.title::text").get(),
                "link": article.css("a::attr(href)").get(),
            }

        # 爬取下一页
        next_page = response.css("a.next::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)