Scrapy Pipeline 管道如何清洗、去重并入库？

Scrapy Pipeline 是 Item 离开 spider 后进入存储系统前的处理链。它适合做数据清洗、字段校验、去重、补全、入库、发送消息等工作。和 spider 相比，Pipeline 更靠近数据出口，所以它不应该关心页面怎么解析，而应该关心“这条数据是否可信、如何保存、失败后怎么办”。

Pipeline 可以配置多个，Scrapy 会按优先级从小到大依次执行。每个 process_item 要么返回 item 交给下一个管道，要么抛出 DropItem 丢弃数据。这里的取舍很实际：把所有逻辑写进一个 Pipeline 简单，但后期很难复用；拆得太细又会增加配置和排查成本。通常可以按职责拆成校验、去重、存储三类。

python
# pipelines.py
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem

class ValidatePipeline:
    def process_item(self, item, spider):
        data = ItemAdapter(item)
        if not data.get('title') or not data.get('url'):
            raise DropItem('missing title or url')
        data['title'] = data['title'].strip()
        return item

class DedupePipeline:
    def open_spider(self, spider):
        self.seen = set()

    def process_item(self, item, spider):
        url = ItemAdapter(item).get('url')
        if url in self.seen:
            raise DropItem(f'duplicate url: {url}')
        self.seen.add(url)
        return item

python
# settings.py
ITEM_PIPELINES = {
    'myproject.pipelines.ValidatePipeline': 100,
    'myproject.pipelines.DedupePipeline': 200,
    'myproject.pipelines.StorePipeline': 300,
}

追问

Pipeline 和 Item Loader 都能清洗数据，应该放哪边？

Loader 更适合处理字段级别的格式问题，比如去空格、取第一个文本、把价格字符串转数字。Pipeline 更适合处理整条数据的业务判断，比如字段是否完整、是否重复、是否需要入库。取舍点是复用范围：页面解析相关的清洗放 Loader，跨 spider 的出口规则放 Pipeline。常见坑是两个地方都清洗同一个字段，最后线上数据异常时没人说得清是哪一步改坏的。

多个 Pipeline 的执行顺序怎么定？

ITEM_PIPELINES 里的数字越小越早执行，所以校验通常放前面，存储放后面。这样缺字段或明显错误的数据可以尽早丢掉，避免浪费数据库连接和网络请求。边界是有些字段需要存储前才生成，比如数据库 ID 或消息队列 trace id，就不能在前置校验里强依赖。实践里优先级不要随手写，最好留出间隔，比如 100、200、300，后面插新步骤更方便。

Pipeline 里连接数据库要注意什么？

连接对象应在 open_spider 中初始化，在 close_spider 中关闭，避免每条 item 都新建连接。每条都建连接会非常慢，还容易把数据库打到连接数上限。取舍是长连接性能好，但要处理断线重连和批量提交失败。边界是小脚本写 CSV 可以简单处理，生产入库则要考虑唯一键、事务、重试和错误日志。

去重应该放内存、Redis 还是数据库？

单机小任务用内存 set 最简单，速度快，也不需要额外服务。多进程、多机器或长周期任务就应该用 Redis 或数据库唯一约束，因为本地 set 无法跨节点共享。取舍是 Redis 去重快但需要维护 key 生命周期，数据库去重可靠但写入压力更大。踩坑点是只用 URL 去重，有些站同一内容存在移动端、PC 端、带追踪参数的多个 URL，需要先做 URL 规范化。

Pipeline 抛出 DropItem 后还会发生什么？

抛出 DropItem 后，这条 item 不会进入后续 Pipeline，Scrapy 会把它记录到 dropped 统计里。这个机制适合丢弃缺字段、重复、明显异常的数据。边界是不要把暂时性错误也直接 Drop，比如数据库超时、接口偶发失败，这类问题更适合重试或落失败队列。否则数据会“安静地丢失”，日志里看得到，但业务侧很难补回来。