做爬虫的同学应该对 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五个字段现在的问题是如何稳定地把数据落到数据库里。我先梳理一下管道的分工管道职责优先级CleanPipeline字段去空格、价格转浮点、日期格式化100DuplicatesPipeline按书名和作者去重200MongoPipeline存入 MongoDB300按照自己在实际项目里的习惯先把需求拆清楚再写代码比一上来就埋头改配置靠谱。清洗管道处理脏数据去重管道保证数据唯一存储管道负责落地每一层只干一件事。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(fDuplicate 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/, serverSelectionTimeoutMS5000 ) 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( hostlocalhost, userroot, passwordyour_password, databasebooks_db, charsetutf8mb4, cursorclasspymysql.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_urisettings.get(MONGO_URI, mongodb://localhost:27017/), mongo_dbsettings.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(hostlocalhost, port6379, db1) 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(fDuplicate item found: {item[book_name]}) self.redis.sadd(self.seen_key, fingerprint) return item这样爬虫挂在服务器上即使中间因为网络原因重启去重数据还在不会重复入库。SISMEMBER 和 SADD 两条命令都是 O(1) 复杂度对爬虫性能影响非常小可以放心大胆用。6. 最后的几条实际心得管道我用了快三年一句话总结它是 Scrapy 里最容易被低估的组件。很多人把它当成“存数据的地方”其实它是整个爬虫数据质量的最后一道闸门。清洗、校验、去重、入库、统计、重试全都可以往里面塞而且每一块都能独立测试、独立调优这是把爬虫项目从“能跑”推到“能稳定跑”的关键一步。实际调试的时候还有一个小技巧写管道初期不要一上来就接数据库先用一个只打印 Item 的管道验证整个链路通不通。跑通之后再一级一级替换成真正的清洗和存储逻辑。这个习惯看起来多了一步实际省的时间非常多尤其是当你同时改多个管道时。如果你正在用 Scrapy 写爬虫并且发现 Spider 越来越臃肿、数据质量越来越难控制就把这篇文章里的管道案例搬过去试试把“解析”和“处理”彻底拆开。你会发现代码结构会干净很多排查问题的效率也会明显上来。后面如果再碰到什么奇怪的管道行为欢迎按这里的思路去排大部分坑都在配置、返回值和资源管理这三个地方。