企业官网建设流程全解析

做爬虫的同学应该对 Scrapy 不陌生。平时写爬虫，很多人习惯在 Spider 里解析数据，然后yield item就完事了，剩下的数据清洗、过滤、入库全靠蜘蛛里自己写的临时代码处理。一旦字段变多、目标站点改版、数据量上来，这套逻辑基本就乱成一锅粥。我第一次用 Scrapy 的时候就是这样，直到某天被一个“某电商网站”的爬虫折腾到凌晨，才认真把 Item Pipeline（管道）从头学了一遍，才发现之前很多问题，根本不是 Scrapy 不行，而是我根本没把管道用明白。

这篇内容就是把 Scrapy 管道从原理到实战完整讲清楚，包括管道在架构里的位置、核心接口、配置顺序、多管道协作、去重、入库、常见坑，以及一些常规文档里不会写的细节。适合刚学完 Scrapy 基础、准备把爬虫写得更“正规”的人，也适合已经写过一两个项目但总觉得数据越爬越乱的初学者。看完能直接照着抄，把你 Spider 里那堆乱七八糟的清洗和存储逻辑，干净地挪到管道里去。

1. 管道到底在 Scrapy 里干活的哪个环节

1.1 一次完整item流动的路线图

先别急着抄代码，把管道的角色搞明白比什么都重要。Scrapy 的数据流可以简单理解成一条流水线：请求从调度器出发，交给下载器拿 HTML，再回到 Spider 解析出数据，最后数据以 Item 对象的形式被 yield 出来。重点是 yield 出来之后呢？很多新手以为到这里爬虫就结束了，其实后面还有引擎、中间件、管道这一大串逻辑等着跑。

管道的官方位置，就是在蜘蛛 yield 出 Item 之后、数据最终落地之前的那一段。它是连接“解析结果”和“数据去向”的中间环节。没有管道，你只能靠 print 看数据、靠文件写成一行一行的人工处理；有了管道，一个 Item 会被自动送进你写好的处理流程里，清洗、校验、去重、存储全在里面完成，蜘蛛本身保持干净清爽。

我在实际项目中，把整条 item 流动分成四步：第一步是 Spider 里构造 Item，第二步是引擎把 Item 交给管道链，第三步是管道按配置顺序依次处理，第四步是数据落库。第四步可能被很多人漏掉，因为 Scrapy 本身不保证数据能自动入库，它只保证把 Item 从一个环节送到下一个环节，最后一步存到哪里、怎么存，完全是管道的事。

1.2 管道的核心接口，不多，就几个

一个标准管道类其实就是一个普通的 Python 类，关键在于它实现了 Scrapy 约定好的几个方法。我用得最多的：process_item、open_spider、close_spider、from_crawler。这里逐个解释一下它们的用途。

process_item(item, spider)是管道里必须实现的方法，每个 Item 流过这个管道时都会调用它一次。这个方法返回 Item 给下一个管道，也可以返回DropItem丢弃当前 Item。从名字就能看出，这是核心中的核心，你在里面写清洗、校验、去重逻辑，做完了把 Item 原样或修改后返回。

open_spider(spider)和close_spider(spider)是生命周期的钩子方法，分别在爬虫启动和关闭时调用一次。适合用来做初始化资源和释放资源的操作，比如建立数据库连接、打开文件、加载历史去重集合，爬虫跑完再去关连接。我在管道里连接 Redis 和 MongoDB 时，基本全靠这两个方法管理资源。

from_crawler(cls, crawler)是类方法，用来从爬虫的全局配置里读取自定义参数，比如数据库地址、用户名、密码，或者某些开关信息。从crawler.settings里取配置，比在管道里硬编码舒服太多了，换环境只需要改配置。

1.3 为什么清洗逻辑别堆积在 Spider 里

很多新手会问：既然 Spider 里也能写清洗和存储，为什么非得用管道？我的回答一般就是：蜘蛛的责任边界是“解析出一个 Item”，管道的责任边界是“把 Item 处理好”。如果你把清洗逻辑写进 Spider，一旦同一个爬虫需要同时跑多个不同站点，或者业务方突然要新增一个字段，你就得在一堆选择器里找那几行处理逻辑。

更重要的是异常处理。Spider 里写循环接存储逻辑，某一条数据格式不对，可能直接让整个爬虫崩溃。而管道天然是一次处理一个 Item，出错时你可以决定记录日志、丢弃还是重试，不会影响其他 Item 的处理流程。数据量上来之后，管道链还能单独拆出来做单元测试，一条 Item 造假数据丢进去，就知道每个环节输出是否正常，比翻蜘蛛日志高效太多了。

2. 管道配置、执行顺序和数据传递

2.1 ITEM_PIPELINES 配置：顺序数字不是摆设

管道写好了，不配置等于没写。启用管道得在settings.py里加一项ITEM_PIPELINES，它的值是一个字典，键是管道类的完整导入路径，值是 0~1000 之间的整数，数字决定了同一批管道相互之间的先后顺序。数字越小越先执行，同数字的执行顺序不保证。

我自己习惯是先做清洗，再做校验去重，最后入数据库。比如你要写三个管道，分别做清洗、去重、入库，配置看起来就是：

ITEM_PIPELINES = { 'myproject.pipelines.CleanPipeline': 100, 'myproject.pipelines.DuplicatesPipeline': 200, 'myproject.pipelines.MongoPipeline': 300, }

间隔为什么要拉大？因为后期很可能要在两个管道之间再插一个流程，如果原来排的是 100 和 101，你插个 150 觉得别扭，插完稍微一改排序就会影响到 URL 指纹和去重逻辑。老老实实隔个几十几十的排，将来好维护，这算是踩过坑之后的一个经验。

2.2 管道协作：Item 不会自动变化，你需要返回它

管道链是串行执行的，前一个管道的输出就是后一个管道的输入。这里有个新手最容易忽略的细节：如果你在process_item里修改了 Item，最后却没有把它 return 出来，下一个管道收到的就是None，后面所有的管道全白写了。

我见过一个特别典型的失误：在清洗管道里把字段字符串去空格、规范日期，然后忘了返回 Item，结果数据库管道一路报“字段不存在”的错。排查了半天，其实就一个 return 丢了。所以建议每个process_item写完后第一件事就是确认它是否返回了 Item 或 DropItem，千万不要只写处理不写返回。

传递数据不只限于 Item 本身。如果想在多个管道之间传递一些临时状态，也可以把它挂在 Item 上，比如item['_skip_flag'] = True，后续管道通过这个标记决定要不要继续处理。这种下划线开头的字段，通常不会进入真正的数据表，注意在入库前删掉即可。

2.3 DropItem 的使用逻辑

DropItem是我在去重场景里用得最多的东西。它在管道里直接触发异常的机制来中断当前 Item 的后续所有管道处理，就是数据被“丢弃”了，后面的存储管道不会拿到这条数据。

举个实际场景：爬到的数据里有重复标题，我本来打算先存库里，再从库里做唯一索引去重。后来想一想，这种做法会让存储管道白忙活，还会产生大量数据库警告。改成在去重管道里做集合判断，遇到重复的立即raise DropItem("duplicate item found: %s" % item["url"])，后面根本不入库存。

唯一的坑是日志。DropItem 默认会按 WARNING 级别输出，如果数据重复率太高，爬虫日志会刷得很难看。所以我实际用的时候，会自己在丢弃前做一次频率统计，超过一定数量再告警，不然一晚上爬几万条数据，日志里全是重复消息，想找真正的问题反而找不到。

3. 实战：从清洗到入库的完整管道链

3.1 场景设计与需求拆解

纸上谈兵没意思，直接拿一个项目来演示。假设我们在爬一个模拟的图书网站（地址用http://books.example.com代替），Spider 已经解析出书名book_name、作者author、价格price、出版日期publish_date、简介description五个字段，现在的问题是如何稳定地把数据落到数据库里。

我先梳理一下管道的分工：

按照自己在实际项目里的习惯，先把需求拆清楚再写代码，比一上来就埋头改配置靠谱。清洗管道处理脏数据，去重管道保证数据唯一，存储管道负责落地，每一层只干一件事。

3.2 清洗管道：字段都能直接用的关键

清洗逻辑看起来简单，真正写起来要注意的细节很多。书本简介里经常有前后空格、乱码符号、价格可能是“¥68.00”这样的长字符串，直接入库后面写报表会疯掉。所以我在清洗管道里统一做类型转换和格式化。

# myproject/pipelines.py import re from datetime import datetime class CleanPipeline: def process_item(self, item, spider): # 去除字符串字段首尾空白 item['book_name'] = item['book_name'].strip() item['author'] = item['author'].strip() if item.get('description'): item['description'] = re.sub(r'\s+', ' ', item['description']).strip() # 价格字段可能带货币符号，先清理再转浮点 price_text = item['price'].replace('￥', '').replace('¥', '').replace(',', '').strip() item['price'] = float(price_text) # 日期字段的格式统一成 yyyy-MM-dd date_text = item['publish_date'].strip() item['publish_date'] = datetime.strptime(date_text, '%Y-%m-%d').date() return item

看起来简单，但有一个经验是“不要假设”字段一定存在。比如老书可能没有简介字段，先get再处理，否则清洗管道会被字段缺失打断。同理，日期格式网上什么奇葩都有，稳妥期间先把格式统一，再交给业务系统去解析。

3.3 去重管道：集合、Redis、还是布隆过滤器

去重是数据爬取里绕不开的话题。同一个站点的列表页可能出现不同入口指向同一本书，你需要决定是保留第一份还是按更新时间更新。我的选择是给每条 Item 生成一个唯一指纹，爬虫启动时加载一遍已有数据的指纹集合，处理时判断指纹是否存在。

import hashlib class DuplicatesPipeline: def open_spider(self, spider): self.seen = set() # 如果是从数据库加载已有数据的指纹，可以在这里做初始化 # self.seen = load_existing_fingerprints() def process_item(self, item, spider): fingerprint = hashlib.md5( (item['book_name'] + item['author']).encode('utf-8') ).hexdigest() if fingerprint in self.seen: raise DropItem(f"Duplicate item found: {item['book_name']}") self.seen.add(fingerprint) return item

这个方案的优点是实现简单，不需要额外组件，单机跑几万条数据毫无压力。缺点是内存会随着 Item 数量增长，几十万条数据以后，集合占用会变得很可观。量级更大时我会换成 Redis 集合做去重，或者用布隆过滤器，能维持一个相对稳定的内存占用。注意布隆过滤器有误判率，会牺牲极少量数据的准确性来换内存，工业数据里通常是可以接受的。

3.4 MongoDB 入库管道：Startup 到 Shutdown 的连接管理

MongoDB 是很多爬虫项目里最顺手的数据仓库，灵活的模式对爬虫这种字段经常变化的数据非常友好。我写 MongoDB 管道的时候，最看重连接资源的管理：爬虫启动时建立连接，爬虫关闭时释放连接，不能每个 Item 都 new 一次客户端，那性能基本废了。

import pymongo class MongoPipeline: def open_spider(self, spider): self.client = pymongo.MongoClient( 'mongodb://localhost:27017/', serverSelectionTimeoutMS=5000 ) self.db = self.client['books_db'] def close_spider(self, spider): self.client.close() def process_item(self, item, spider): self.db['books'].insert_one(dict(item)) return item

这里有几个隐蔽的细节。第一，如果清洗管道把某些字段转成了日期类型，MongoDB 是支持直接存日期的，查询时做区间过滤很方便，不要为了省事把日期存成字符串，不然将来统计时你会后悔。第二，如果同一个 Item 在 Mongo 管道里被 insert 了，然后你继续返回给它之后的管道，没有问题；但如果在它后面还有一个搜索索引管道，记得它已经在库里了，索引管道的逻辑要自己判断。第三，千万要在close_spider里close()，不然连接一直挂着，本地开发还好，线上部署时会焊死一堆连接数。

3.5 MySQL 入库管道：写入失败不能静默

虽然 MongoDB 用得多，但业务部门往往更认 MySQL，毕竟 BI 工具、报表系统基本都离不开关系数据库。用管道写 MySQL，核心思路是一样，但细节更多。我一般不在管道里拼动态 SQL 去 insert，而是把重复的字段组成一个元组，批量执行。

import pymysql class MySQLPipeline: def open_spider(self, spider): self.connection = pymysql.connect( host='localhost', user='root', password='your_password', database='books_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) self.cursor = self.connection.cursor() def close_spider(self, spider): self.cursor.close() self.connection.close() def process_item(self, item, spider): sql = """ INSERT INTO books (book_name, author, price, publish_date, description) VALUES (%s, %s, %s, %s, %s) """ values = ( item['book_name'], item['author'], item['price'], item['publish_date'], item.get('description', '') ) self.cursor.execute(sql, values) self.connection.commit() return item

这里我最想提醒的是：不要每一条数据都 commit。一条条 commit 在数据量小的时候感觉不到，一旦爬到几万条，每条都 commit 一次，性能会掉得非常明显。更合理的做法是在管道里攒一批，比如攒满 200 条或者close_spider时统一 commit。但注意如果中途爬虫崩溃，没提交的数据会丢，所以得在业务容忍度之间找一个平衡点。

4. 管道常见问题与排查经验

4.1 管道没执行，八成是配置或导入路径的问题

新手遇到最多的情况是：Spider 明明 yield 了数据，但管道里的 print 一条都不打印。这个问题的排查顺序一般是先看settings.py里ITEM_PIPELINES是否写对，路径和类名是否和文件里完全一致，包括大小写。

其次看 Spider 里 yield 的对象是不是真正的 Item 对象。如果你 yield 的是字典{}，管道的process_item接口也能收到，但某些依赖 Item 字段定义的方法会报错。再说直白点，如果你把 Spider 多写了一个 yield 的 return，也可能出现管道不处理它的情况。所以排查时先建一个最简单的管道，只打印一句话，从简到繁，别上来就检查去重算法。

还有个坑：Scrapy 启动时如果ITEM_PIPELINES配置为空字典，你的管道类即使写了也不会被加载。这个空配置和“没配置”等价，很容易被当成环境不一致忽视掉。

4.2 process_item 的返回值丢了，导致链路中断

上一个管道没返回 Item，下一个管道的入参就是None。你在后续管道里写的item['field']直接抛TypeError或者KeyError。这算是我看求助帖时出现频率极高的一个问题。

解法也简单：在每个管道的process_item写好之后，加一行logger.debug或者print，把收到的 item 和返回的 item 关键字段打出来，跑一轮看链路数据，两步就能定位是谁吃掉了 Item。别嫌麻烦，爬到后面问题越藏越深，前期多打一两个日志能省下大把时间。

4.3 去重集合内存暴涨怎么办

用 Python 集合去重，随着 Item 数量增加内存暴涨是必然的。我试过爬一个有几十万数据的站点，爬了一大半进程就 OOM 了。后来改成在每个管道处理的 Item 量超过一定阈值时，定期把集合同步到 Redis 并清理一部分历史数据，内存压力才降下来。

如果项目要求不高，也可以做个简单方案：基于 Redis 的SADD和SISMEMBER，几行代码搞定；或者用布隆过滤器，靠谱、内存小，缺点是安装依赖多一个。具体选型看数据量：十万以下用内存集合就行，百万级用 Redis set，千万级再考虑布隆过滤，别一开始就上重武器。

4.4 数据库写入异常不能中断整次爬虫

管道里直接写数据库，最怕的就是网络闪断、数据结构变化导致入库报错。如果只是偶尔一条脏数据，你的爬虫被中断，前面的努力全白费。我的做法是在入库管道里做异常隔离：

class MongoPipeline: def process_item(self, item, spider): try: self.db['books'].insert_one(dict(item)) except pymongo.errors.DuplicateKeyError: spider.crawler.stats.inc_value('database/duplicate') except pymongo.errors.ServerSelectionTimeoutError: spider.crawler.stats.inc_value('database/error') return item

注意这里我们捕获了 DuplicateKeyError 不代表要丢失数据，有可能是唯一索引命中了，直接记一条统计就行。对于网络超时这种错误，合理的策略是重试几次，如果连续失败再抛异常，让整个爬虫停下来。否则无限重试也会变成“假死”，看起来还在跑，实际一条数据都没入。

4.5 管道顺序调整后，行为完全变了

有次我调整了清洗管道和去重管道的顺序，结果去重管道拿到的还是带有空格的书名，去重率直接降低，库里出现一堆“相同”但清洗前后不一样的书。这段经历告诉我一个原则：去重之前必须先把字段清洗干净，否则指纹算的是脏数据，清洗后的数据被当成新数据存了进去。

另一个容易被忽略的点是，如果你在清洗管道里把价格从字符串转成了浮点数，那么后续管道和数据库里看到的都是浮点数。如果你在去重管道前还写了一个把价格格式化的管道，顺序没排好就可能两边不一致。所以管道顺序不要随意动，动了以后至少测一轮“从 yield 到入库”的完整数据流再上线。

5. 管道的几个进阶用法

5.1 用 from_crawler 读取自定义配置

管道类里写死 MongoDB 连接地址，换环境就要改代码，这是很糟糕的维护状态。更好的是把配置放到settings.py里，用from_crawler来取。这样本地开发、测试、线上部署只用改配置，不用动管道代码。

class MongoPipeline: @classmethod def from_crawler(cls, crawler): settings = crawler.settings return cls( mongo_uri=settings.get('MONGO_URI', 'mongodb://localhost:27017/'), mongo_db=settings.get('MONGO_DB', 'books_db') ) def __init__(self, mongo_uri, mongo_db): self.client = pymongo.MongoClient(mongo_uri) self.db = self.client[mongo_db]

这里重点说一下为什么用from_crawler而不是直接在__init__里get_settings。因为process_item的签名会限制你直接访问 settings，而你通过from_crawler可以非常优雅地在类初始化时把配置注入进去，这个写法也是 Scrapy 官方推荐的，阅读代码时别人一眼就能明白这个管道依赖了哪些配置项。

5.2 攒批入库，把性能提升一个量级

如果数据量真的很大，单条一条入库会成为瓶颈。我曾经遇到一个项目，每秒钟要入几十条数据，MongoDB 单条insert_one明显扛不住，后来改成在管道里攒批，攒到 100 条用insert_many一下写进去，性能直接翻倍。

class BatchMongoPipeline: BATCH_SIZE = 100 def open_spider(self, spider): self.client = pymongo.MongoClient('mongodb://localhost:27017/') self.db = self.client['books_db'] self.buffer = [] def process_item(self, item, spider): self.buffer.append(dict(item)) if len(self.buffer) >= self.BATCH_SIZE: self.db['books'].insert_many(self.buffer) self.buffer.clear() return item def close_spider(self, spider): if self.buffer: self.db['books'].insert_many(self.buffer) self.buffer.clear() self.client.close()

攒批最关键的是close_spider时一定要把残留的不足一批的数据写进去，否则最后一批数据会静默丢失。这个错误我犯过一次，深刻教训，不写出来你们可能也会踩。

5.3 管道里做状态统计

管道不光是“处理 Item”，它还是天然的监控点。我喜欢在管道里统计清洗前后的字段数量、丢弃了多少、入库成功多少。Scrapy 自带的spider.crawler.stats能很方便地累加这些计数，爬完导出数据就是一份现成的质量报告。

def process_item(self, item, spider): spider.crawler.stats.inc_value('pipeline/clean/received') # ... 清洗逻辑 ... spider.crawler.stats.inc_value('pipeline/clean/done') return item

在调试阶段，这套统计比你的 print 日志有用得多。你可以在爬虫跑完后直接看stats的输出，清晰看到哪一步耗时最多、哪一步丢数据最严重。把统计打印出来，往往几个数字就能暴露问题。我自己在做一个长时间爬虫时，也会在管道里周期性打印当前入库总数，免得看着爬虫跑了一个小时心里没底。

5.4 和爬虫暂停/断点续爬配合

不少人会问：爬虫停了重启，管道里的去重集合不就丢了吗？如果去重状态只存在内存里，重启后确实会丢。所以做严肃项目时，我通常会把指纹集合写入 Redis，每次open_spider时加载，close_spider时保存，或者至少在每次有新指纹时同步到 Redis。

class RedisDuplicatesPipeline: def open_spider(self, spider): self.redis = redis.Redis(host='localhost', port=6379, db=1) self.seen_key = f"{spider.name}:seen" def process_item(self, item, spider): fingerprint = hashlib.md5( (item['book_name'] + item['author']).encode('utf-8') ).hexdigest() if self.redis.sismember(self.seen_key, fingerprint): raise DropItem(f"Duplicate item found: {item['book_name']}") self.redis.sadd(self.seen_key, fingerprint) return item

这样爬虫挂在服务器上，即使中间因为网络原因重启，去重数据还在，不会重复入库。SISMEMBER 和 SADD 两条命令都是 O(1) 复杂度，对爬虫性能影响非常小，可以放心大胆用。

6. 最后的几条实际心得

管道我用了快三年，一句话总结：它是 Scrapy 里最容易被低估的组件。很多人把它当成“存数据的地方”，其实它是整个爬虫数据质量的最后一道闸门。清洗、校验、去重、入库、统计、重试，全都可以往里面塞，而且每一块都能独立测试、独立调优，这是把爬虫项目从“能跑”推到“能稳定跑”的关键一步。

实际调试的时候还有一个小技巧：写管道初期不要一上来就接数据库，先用一个只打印 Item 的管道验证整个链路通不通。跑通之后，再一级一级替换成真正的清洗和存储逻辑。这个习惯看起来多了一步，实际省的时间非常多，尤其是当你同时改多个管道时。

如果你正在用 Scrapy 写爬虫，并且发现 Spider 越来越臃肿、数据质量越来越难控制，就把这篇文章里的管道案例搬过去试试，把“解析”和“处理”彻底拆开。你会发现代码结构会干净很多，排查问题的效率也会明显上来。后面如果再碰到什么奇怪的管道行为，欢迎按这里的思路去排，大部分坑都在配置、返回值和资源管理这三个地方。