Scrapy Item Pipeline 完全指南：编写、启用与调试 Item 处理组件

Item Pipeline 是 Scrapy 中处理抓取结果的核心机制：spider 产出的每个 item 都会按固定顺序流经若干 pipeline 组件，由这些组件完成清洗、校验、去重、落库等操作。本文基于 Item Pipeline 官方文档 与当前仓库源码，完整覆盖组件的编写方式（process_item / open_spider / close_spider）、五个实战示例、ITEM_PIPELINES 配置规则，以及调试命令与常见坑位，读完后你可以独立写出生产级 pipeline 并用 scrapy parse 对其进行验证。

Item Pipeline 的定位与执行模型

一个 item 被 spider 抓取之后，会交给 Item Pipeline 处理，并由多个组件顺序执行：每个 pipeline 组件都是一个实现了简单方法的 Python 类，它接收 item、对其执行操作，同时决定该 item 是继续流向下一个组件，还是被丢弃（drop）后不再处理。

Item pipeline 的典型用途包括：

• 清洗 HTML 数据；
• 校验抓取结果（检查 item 是否包含某些必需字段）；
• 查重（发现重复后直接丢弃）；
• 将 item 存储到数据库。

源码视角下的调度顺序

从源码结构看，pipeline 的调度由 ItemPipelineManager 完成，它复用 Scrapy 的中间件管理基础设施（MiddlewareManager）：

• process_item：各组件依次串行执行，前一个组件的输出是后一个组件的输入；
• open_spider / close_spider：各组件并行调用（源码中通过 _process_parallel 并发执行），且 close_spider 以 appendleft 方式插入——这意味着关闭阶段与打开阶段的组件顺序相反，先初始化的组件最后释放资源；
• 组件列表由 ITEM_PIPELINES 设置中的优先级数字排序生成（build_component_list 解析 settings 中的 ITEM_PIPELINES）。

item 处理过程中的异常语义由 Scraper.start_itemproc_async 定义：

• 组件抛出 DropItem → 记录 drop 日志并发送 item_dropped 信号，item 终止流程；
• 组件抛出其他异常 → 记录错误日志并发送 item_error 信号，item 同样被放弃；
• 正常返回 → 发送 item_scraped 信号，item 进入 feed export 或其他输出。

编写你自己的 Item Pipeline

process_item（必须实现）

每个 item pipeline 组件必须实现 process_item(self, item) 方法。该方法对每个 item 都会被调用：

• item 是 item 对象，可以是 scrapy.Item、dict 或任意支持 ItemAdapter 的类型；
• process_item 必须返回 item 对象，或抛出 DropItem 异常；
• 被 drop 的 item 不再被后续组件处理。

注意 DropItem 支持一个可选参数 log_level（见 scrapy/exceptions.py），可用于调整 drop 日志的记录级别：

raise DropItem("Missing price", log_level=logging.WARNING)

open_spider / close_spider（可选）

组件还可以实现以下两个生命周期方法：

• open_spider(self)：spider 打开时被调用。从 2.18.0 起，它可以抛出 CloseSpider 异常，在爬虫开始抓取前直接关闭爬虫——例如 pipeline 所依赖的资源不可用时（文档中的 versionchanged 说明）；
• close_spider(self)：spider 关闭时被调用，且发生在 spider_closed 信号发出之前。

以上任意方法都可以定义为协程函数（async def），Scrapy 会正确等待其完成。新建项目时，startproject 模板 已经生成了一份 pipelines.py 骨架，其中明确提醒："Don't forget to add your pipeline to the ITEM_PIPELINES setting"——这正是后面“常见坑位”一节讨论的第一个问题。

实战示例（文档原样继承并注释）

价格校验与丢弃无价格 item

下面的 pipeline 对不含 VAT（price_excludes_vat 属性）的 item 调整 price 属性，并丢弃没有价格的 item：

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


class PricePipeline:
    vat_factor = 1.15

    def process_item(self, item):
        adapter = ItemAdapter(item)
        if adapter.get("price"):
            if adapter.get("price_excludes_vat"):
                adapter["price"] = adapter["price"] * self.vat_factor
            return item
        else:
            raise DropItem("Missing price")

要点：用 ItemAdapter 统一访问 item 属性，兼容 scrapy.Item / dict / dataclass 等多种 item 类型；get 返回 None 时走 DropItem 分支。

将 item 写入 JSON Lines 文件

下面的 pipeline 将所有 spider 抓取到的 item 写入单个 items.jsonl 文件，每行一个 item 的 JSON 序列化：

import json

from itemadapter import ItemAdapter


class JsonWriterPipeline:
    def open_spider(self):
        self.file = open("items.jsonl", "w")

    def close_spider(self):
        self.file.close()

    def process_item(self, item):
        line = json.dumps(ItemAdapter(item).asdict()) + "\n"
        self.file.write(line)
        return item

官方文档特别注明：JsonWriterPipeline 的目的只是演示如何写 item pipeline。如果真的要“把所有 item 存到 JSON 文件”，应使用 Feed exports 功能，而不是自己写文件——后者没有批处理、分片与失败重试等工程化能力。

将 item 写入 MongoDB

这个示例展示如何拿到 crawler 对象、以及如何正确释放资源。MongoDB 地址与库名来自 Scrapy settings，集合名由类属性指定：

import pymongo
from itemadapter import ItemAdapter


class MongoPipeline:
    collection_name = "scrapy_items"

    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_DATABASE", "items"),
        )

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

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

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

from_crawler 是组件的标准入口：Scrapy 实例化 pipeline 时会优先调用该类方法，从而让组件从 crawler.settings 读取配置（MONGO_URI、MONGO_DATABASE，后者缺省为 "items"）。连接在 open_spider 中建立、close_spider 中关闭，是资源管理的标准范式。

对 item 截图（协程版 process_item）

这个示例演示在 process_item 中使用 协程语法：pipeline 向本地运行的 Splash 实例发起请求，渲染 item URL 的截图，保存截图文件并把文件名写回 item：

import hashlib
from pathlib import Path
from urllib.parse import quote

import scrapy
from itemadapter import ItemAdapter
from scrapy.http.request import NO_CALLBACK


class ScreenshotPipeline:
    """Pipeline that uses Splash to render screenshot of
    every Scrapy item."""

    SPLASH_URL = "http://localhost:8050/render.png?url={}"

    def __init__(self, crawler):
        self.crawler = crawler

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler)

    async def process_item(self, item):
        adapter = ItemAdapter(item)
        encoded_item_url = quote(adapter["url"])
        screenshot_url = self.SPLASH_URL.format(encoded_item_url)
        request = scrapy.Request(screenshot_url, callback=NO_CALLBACK)
        response = await self.crawler.engine.download_async(request)

        if response.status != 200:
            # Error happened, return item.
            return item

        # Save screenshot to file, filename will be hash of url.
        url = adapter["url"]
        url_hash = hashlib.md5(url.encode("utf8")).hexdigest()
        filename = f"{url_hash}.png"
        Path(filename).write_bytes(response.body)

        # Store filename in item.
        adapter["screenshot_filename"] = filename
        return item

注意两处细节：async def process_item 通过 await self.crawler.engine.download_async(request) 发起真实的下载，且 callback=NO_CALLBACK 表示该请求不进入 spider 的回调流程；出错时（状态码非 200）pipeline 选择返回原 item 而非丢弃，让后续流程不受截图失败影响。

重复 item 过滤器

假设 item 有唯一 id，但 spider 可能返回多个相同 id 的 item，下面的过滤器会丢弃已处理过的 item：

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


class DuplicatesPipeline:
    def __init__(self):
        self.ids_seen = set()

    def process_item(self, item):
        adapter = ItemAdapter(item)
        if adapter["id"] in self.ids_seen:
            raise DropItem(f"Item ID already seen: {adapter['id']}")
        else:
            self.ids_seen.add(adapter["id"])
            return item

启用 pipeline 组件：ITEM_PIPELINES 设置

要让一个 pipeline 组件生效，必须把它的类写入 ITEM_PIPELINES 设置（通常在项目的 settings.py 中）：

ITEM_PIPELINES = {
    "myproject.pipelines.PricePipeline": 300,
    "myproject.pipelines.JsonWriterPipeline": 800,
}

• 键是组件类的全限定 import 路径；
• 值为整数优先级，item 从数值小的流向数值大的组件（PricePipeline 先于 JsonWriterPipeline 执行）；
• 惯例上这些数字取 0–1000 范围。

启用后，爬虫启动日志中会出现一行由 scrapy.middleware 输出确认的信息：

[scrapy.middleware] INFO: Enabled item pipelines:
['myproject.pipelines.PricePipeline']

一个完整示例：四件套如何协同

上面的例子都是孤立的组件。在真实项目中，pipeline 是四个协同部件之一：spider 产出的 item、yield 它的 spider、处理它的 pipeline，以及启用它的 ITEM_PIPELINES 设置。下面以 books.toscrape.com 为例把四者接起来，复用前文的 PricePipeline。

第一步，在 myproject/items.py 中定义 item（dataclass 形式的 item 是 Scrapy 当前支持的标准 item 类型之一）：

from dataclasses import dataclass


@dataclass
class BookItem:
    title: str
    price: float

第二步，在 myproject/spiders/books.py 的 spider 中 yield item 实例：

import scrapy

from myproject.items import BookItem


class BooksSpider(scrapy.Spider):
    name = "books"
    start_urls = ["https://books.toscrape.com/"]

    def parse(self, response):
        for book in response.css("article.product_pod"):
            yield BookItem(
                title=book.css("h3 a::attr(title)").get(),
                price=float(book.css("p.price_color::text").re_first(r"[\d.]+")),
            )

第三步，把前文的 PricePipeline 放入 myproject/pipelines.py，并在 myproject/settings.py 中启用：

ITEM_PIPELINES = {
    "myproject.pipelines.PricePipeline": 300,
}

这样，BooksSpider yield 的每个 BookItem 都会先经过 PricePipeline 校验与调价，然后才到达 feed exports 或其他输出。

测试 Item Pipeline：scrapy parse --pipelines

如果只想把单个 URL 的 item 送入 pipeline 验证，而不跑完整爬虫，可以使用 parse 命令加 --pipelines 选项：

scrapy parse --pipelines "https://books.toscrape.com/"

如果要针对特定 item 数据做测试，可以给 spider 添加一个“从关键字参数构建 item”的回调：

class BooksSpider(scrapy.Spider):
    # ...

    def parse_item(self, response, **fields):
        yield BookItem(**fields)

然后在命令行传入关键字参数：

scrapy parse --pipelines -c parse_item --cbkwargs '{"title": "Test", "price": 10}' "https://books.toscrape.com/"

文档提示：URL 传 spider 能处理的任意地址即可，虽然回调会忽略该响应，它仍会被真实下载——这是 parse 命令的工作方式。

常见坑位

坑位一：pipeline 根本没运行

pipeline 组件只有在其类被列入 ITEM_PIPELINES 设置（通常位于项目 settings.py）时才会运行；把它写到 spider 类里或其他任何地方都无效。

确认 Scrapy 是否加载了 pipeline：查看爬虫日志开头附近是否有

[scrapy.middleware] INFO: Enabled item pipelines:
['myproject.pipelines.PricePipeline']

如果列表中没有你的 pipeline，检查两点：

1. 类的全限定 import 路径与 ITEM_PIPELINES 中的条目完全一致；
2. 设置没有被覆盖——例如被 spider 的 custom_settings 属性覆盖，或被 settings.py 中重复定义的 ITEM_PIPELINES 覆盖。

坑位二：item 没有被返回

process_item 必须返回 item（或抛出 DropItem）。最常见的错误是修改了 item 却忘记 return：

def process_item(self, item):
    ItemAdapter(item)["price"] *= 1.15
    # Bug: returns None, so the next component gets None instead of the item.

正确写法：

def process_item(self, item):
    ItemAdapter(item)["price"] *= 1.15
    return item

只有把 item 返回，后续组件和 Scrapy 的其余部分才能继续处理它。

小结

Item Pipeline 的接口很小（一个必选方法 + 两个生命周期钩子），但调度语义值得记牢：process_item 串行、生命周期方法并行且 close_spider 逆序执行、DropItem 立即终止 item 流程并触发 item_dropped 信号（见 ItemPipelineManager 与 Scraper.start_itemproc_async）。配合 ITEM_PIPELINES 的优先级排序、scrapy parse --pipelines 的单 URL 调试，以及日志中 "Enabled item pipelines" 的加载确认，足以覆盖日常开发中 pipeline 的编写、启用、验证与排障全链路。