Scrapy实战(3)：启用数据管道清洗行情

Scrapy实战(3)：启用数据管道清洗行情

小象量化极客小册 · 华尔街量化分析先锋系列

本节我们将探寻如何使用 Scrapy 框架进行网络数据采集。作为量化分析师，你将编写爬虫从财经网站爬取最新的股票行情和社交媒体上的咖啡店舆情，为策略引擎输入第一手另类数据。

小象量化极客小册 · 华尔街量化分析先锋系列 本节我们将探寻如何使用 Scrapy 框架进行网络数据采集。作为量化分析师，你将编写爬虫从财经网站爬取最新的股票行情和社交媒体上的咖啡店舆情，为策略引擎输入第一手另类数据。

💡 核心导读

本节将深入Scrapy框架的核心工作流程，重点讲解Item Pipeline的数据处理机制、中间件(Middleware)的定制方法以及项目优化技巧。通过完整的爬虫案例演示如何清洗存储数据、处理异常请求，并实现分布式爬取。掌握这些技能后，读者将能够构建工业级的数据采集系统。

数据处理管道(Item Pipeline)

管道工作机制

Scrapy的Item Pipeline是数据处理流水线，包含多个按序执行的处理器：

爬虫解析 → Pipeline1 → Pipeline2 → ... → 持久化存储

典型管道组件实现

import pymongo
from itemadapter import ItemAdapter

class PriceConverterPipeline:
    """将价格字段统一转换为美元"""
    exchange_rate = 7.2  # 人民币兑美元汇率

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if 'price_cny' in adapter:
            adapter['price_usd'] = round(adapter['price_cny'] / self.exchange_rate, 2)
            del adapter['price_cny']
        return item

class MongoDBPipeline:
    """存储到MongoDB数据库"""
    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DB')
        )

    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]

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

    def process_item(self, item, spider):
        self.db[spider.name].insert_one(ItemAdapter(item).asdict())
        return item

管道启用配置

ITEM_PIPELINES = {
    'project.pipelines.PriceConverterPipeline': 100,  # 数值越小优先级越高
    'project.pipelines.MongoDBPipeline': 200,
}
MONGO_URI = 'mongodb://localhost:27017'
MONGO_DB = 'scrapy_data'

💡 概念小测：Pipeline流转

在 Scrapy 项目中，如果一个 Item 经过了一个数据管道 PriceCleanerPipeline，并且在它的 process_item 方法中没有 return item，会导致什么后果？

• 这个 Item 会被自动存入本地 JSON 文件。
• 该 Item 会从数据流中消失，后续排序更后的 Pipeline 将无法接收到它。
• Scrapy 引擎会立刻崩溃并抛出 SyntaxError 异常。
• 爬虫会重新爬取该 Item 对应的网页。

📝 选择题

💡 概念小测：Pipeline流转 在 Scrapy 项目中，如果一个 Item 经过了一个数据管道 PriceCleanerPipeline，并且在它的 process_item 方法中没有 return item，会导致什么后果？

• A. 这个 Item 会被自动存入本地 JSON 文件。
• B. 该 Item 会从数据流中消失，后续排序更后的 Pipeline 将无法接收到它。
• C. Scrapy 引擎会立刻崩溃并抛出 SyntaxError 异常。
• D. 爬虫会重新爬取该 Item 对应的网页。

答案：B（该 Item 会从数据流中消失，后续排序更后的 Pipeline 将无法接收到它。）

中间件开发实战

下载中间件示例

from scrapy import signals
from scrapy.downloadermiddlewares.userpe_rationt import UserAgentMiddleware
import random

class RandomUserAgentMiddleware(UserAgentMiddleware):
    """随机User-Agent中间件"""
    user_pe_rationts = [
        'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
        'AppleWebKit/537.36 (KHTML, like Gecko)',
        'Chrome/91.0.4472.124 Safari/537.36'
    ]

    def process_request(self, request, spider):
        request.headers['User-Agent'] = random.choice(self.user_pe_rationts)

中间件配置激活

DOWNLOADER_MIDDLEWARES = {
    'project.middlewares.RandomUserAgentMiddleware': 543,
}

分布式爬虫实现

Redis分布式配置

SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://:password@localhost:6379'

数据去重对比

原始数据:
+----+----------------------------+--------+
| id |           title            | price  |
+----+----------------------------+--------+
| 1  | Python编程指南             | 59.99  |
| 1  | Python编程指南             | 59.99  | ← 重复数据
| 2  | 数据科学实战               | 79.50  |
+----+----------------------------+--------+

去重后:
+----+----------------------------+--------+
| id |           title            | price  |
+----+----------------------------+--------+
| 1  | Python编程指南             | 59.99  |
| 2  | 数据科学实战               | 79.50  |
+----+----------------------------+--------+

💡 代码填空：启用管道

补全项目 settings.py 中激活自定义行情清洗管道的代码：

ITEM_PIPELINES = {
    'myproject.pipelines.StockCleanPipeline': {blank},
}

• True
• ‘active’
• 300

✏️ 填空题

💡 代码填空：启用管道 补全项目 settings.py 中激活自定义行情清洗管道的代码：

# settings.py 配置
ITEM_PIPELINES = {
    'myproject.pipelines.StockCleanPipeline': ___,
}

答案：300

本章小结

行动清单

• 实现一个自定义Pipeline，将采集的日期字段统一转换为ISO8601格式
• 开发下载中间件，自动处理遇到403状态码时切换代理IP
• 配置Redis分布式环境，验证多节点协同爬取效果

「小象量化教研组」