Scrapy+MongoDB数据存储全攻略：从Pipeline到分布式爬虫实践

做爬虫时间长了就会发现真正折磨人的往往不是反爬、不是解析而是数据存下来之后怎么管、怎么用。早期我用Scrapy写爬虫最常干的事就是把数据一股脑丢进CSV或者MySQL结果要么是字段对不上要么是表结构改起来要命后来项目里换了MongoDB做存储搭配Scrapy的Pipeline机制整个数据链路一下子顺了很多。这篇文章就是围绕“scrapy框架MongoDB保存”这条主线把我实际踩过的坑、验证过的方案、还有那些文档里不会明说的细节完整梳理一遍给正在做爬虫存储选型或者卡在数据落库环节的朋友一个可以直接抄作业的参考。1. 为什么爬虫存储我最终选了MongoDB1.1 爬虫数据的存储困境很多初学者第一次写爬虫数据量小的时候随便存个CSV就完事了。但一旦爬虫跑起来几天不出问题数据量涨到几百万条CSV的短板就全暴露了并发写入锁冲突、字段追加要改全表、按条件查询慢得离谱更别提数据一旦粘连格式错误整份文件可能直接报废。MySQL这类关系型数据库确实是绝大多数项目的默认选择但在爬虫场景里它有不少别扭的地方。爬虫抓回来的数据结构天生就是“半结构化”的——今天页面多了个字段明天某个字段是嵌套的列表后天又冒出一个JSON对象用MySQL就得提前设计好表结构字段一变就得ALTER TABLE跑批爬到一半改表结构简直是噩梦。再加上多张表之间的关联查询在爬虫项目里绝大多数时候根本用不上。1.2 MongoDB在爬虫场景的核心优势MongoDB是文档型数据库存的是BSON格式本质上是二进制化的JSON。这意味着爬虫解析出来的数据长什么样存进去就什么样不需要做对象关系映射不需要设计表结构。我自己最直观的感受是三个字省事、快、灵活。第一是省事。Scrapy的Item解析出来是dict结构直接交给MongoDB的insert_one方法就能落库字段自动变成文档的key不用写建表语句不用管字段长度和类型后续要加字段就直接加老数据没有这个字段也无所谓根本不存在“改表结构”这件事。第二是快。MongoDB的写入性能在非事务场景下远优于MySQL配合批量插入几万条数据也就几秒钟的事。爬虫本身就是高吞吐写入场景MongoDB的写入模型天然契合。第三是灵活。爬取过程中经常遇到同一个Item在不同页面字段数量不一样的情况MongoDB允许同一集合里的文档拥有完全不同的字段结构这在反爬策略频繁变化、页面结构经常调整的实战中太重要了。1.3 什么时候应该放弃MongoDBMongoDB不是万能的如果项目要求强事务一致性比如订单、余额这类涉及资金的数据那老老实实用MySQL或者PostgreSQL别拿爬虫的经验硬套。还有如果业务上需要大量的多表关联查询MongoDB的关联能力很弱虽然能用$lookup模拟但性能和复杂度都不划算。另外要注意一点MongoDB默认没有开启认证如果部署在公网服务器上极容易被勒索攻击这类事件太多了。要么用内网部署要么必须开启认证并限制IP白名单。这块我在后面踩坑部分会详细说。提示爬虫数据存储选型核心判断标准就两条——数据结构是否多变、写入并发是否高。如果两个都占MongoDB几乎是最优解如果数据强结构化且需要复杂事务老老实实用关系型数据库。2. 环境准备从零搭好Scrapy和MongoDB2.1 MongoDB安装与基础配置先说安装。我自己主力开发机是Windows服务器是Debian两边的安装方式都踩过一遍。Windows下安装MongoDB很简单去官网下载MSI安装包一路Next就行。需要注意的是安装完成后要手动把C:\Program Files\MongoDB\Server\6.0\bin加进系统PATH否则命令行里找不到mongod命令。安装服务的时候记得勾选“Install MongoDB as a Service”这样开机自动启动省心。Debian服务器上安装很多人会直接apt install mongodb但这装的是过时的版本且已经被移出官方源了。正确做法是先导入MongoDB官方GPG公钥再添加官方源然后安装。我习惯用一套现成的命令curl -fsSL https://www.mongodb.org/static/pgp/server-6.0.asc | \ sudo gpg -o /usr/share/keyrings/mongodb-server-6.0.gpg --dearmor echo deb [ signed-by/usr/share/keyrings/mongodb-server-6.0.gpg ] http://repo.mongodb.org/apt/debian bullseye/mongodb-org/6.0 main | \ sudo tee /etc/apt/sources.list.d/mongodb-org-6.0.list sudo apt-get update sudo apt-get install -y mongodb-org安装完先别急着跑业务建议把认证打开。启动服务后进入mongo shell创建管理员账号use admin db.createUser({ user: root, pwd: your_strong_password, roles: [{ role: root, db: admin }] })然后编辑/etc/mongod.conf把security.authorization改成enabled重启服务。这一步在部署到公网服务器时是保命操作千万别省略。2.2 创建Scrapy项目与目录结构环境就绪后用命令创建项目scrapy startproject myproject cd myproject scrapy genspider example example.com生成的目录结构里最核心的四个文件是items.py、pipelines.py、settings.py和spiders/下的爬虫文件。很多新手分不清这几个文件的职责我打个比方爬虫文件是生产线上的工人负责从网页上抓取原始材料items.py是产品规格书定义材料怎么组织成标准件pipelines.py是质检和仓储环节负责把标准件清洗、加工、最后入库settings.py是工厂管理制度控制整条线的运行参数。2.3 Items定义数据结构的起点写爬虫项目我习惯先写Items哪怕爬虫还没写一行代码先把数据结构定下来。这样后面写解析逻辑和Pipeline的时候有据可依不会东一榔头西一棒子。import scrapy class ProductItem(scrapy.Item): product_id scrapy.Field() name scrapy.Field() price scrapy.Field() description scrapy.Field() category scrapy.Field() images scrapy.Field() # 列表 specs scrapy.Field() # 嵌套字典 crawl_time scrapy.Field()注意images和specs这两个字段一个是列表一个是嵌套字典这在MongoDB里对应数组和内嵌文档非常自然。如果换成MySQL这两个字段就得单独建表或者用JSON字符串存查询和清洗都麻烦得多。这是我推荐MongoDB存储最直接的理由之一。注意Items里的字段命名尽量统一风格我习惯全小写下划线。有些人喜欢驼峰命名后面写MongoDB聚合查询时JavaScript语法的字段名和Python风格混着来极其容易出错。3. Pipeline深度实现数据落库全流程3.1 Pipeline基础结构与连接管理Scrapy的Pipeline核心就是一个process_item方法每个Item从爬虫出来后会按优先级依次经过所有启用的Pipeline。先写一个最小可用的MongoDB Pipelineimport pymongo class MongoPipeline: def __init__(self, mongo_uri, mongo_db, mongo_collection): self.mongo_uri mongo_uri self.mongo_db mongo_db self.mongo_collection mongo_collection self.client None self.db None self.collection None classmethod def from_crawler(cls, crawler): return cls( mongo_uricrawler.settings.get(MONGO_URI, mongodb://localhost:27017), mongo_dbcrawler.settings.get(MONGO_DB, scrapy_data), mongo_collectioncrawler.settings.get(MONGO_COLLECTION, items) ) def open_spider(self, spider): self.client pymongo.MongoClient(self.mongo_uri) self.db self.client[self.mongo_db] self.collection self.db[self.mongo_collection] def close_spider(self, spider): self.client.close() def process_item(self, item, spider): self.collection.insert_one(dict(item)) return item很多人写Pipeline会把MongoClient直接写在模块顶部初始化这在Scrapy里其实是有问题的。Scrapy的spider可以动态创建Pipeline实例化时机和数量都不确定如果在模块级别创建MongoClient会出现连接句柄泄漏、连接池爆满的问题。正确做法是像上面这样用from_crawler读取配置在open_spider时创建连接在close_spider时关闭生命周期与爬虫运行周期绑定。3.2 去重与增量更新爬虫最怕重复数据页面的分页逻辑一旦有交叉或者反爬让你重复请求入库的数据就全是重复的。我早期吃过这个亏几百万条数据里有将近一半是重复的想清理还得写脚本慢慢筛。MongoDB天然支持唯一索引这是去重最干净利落的方式。给product_id字段建唯一索引def open_spider(self, spider): self.client pymongo.MongoClient(self.mongo_uri) self.db self.client[self.mongo_db] self.collection self.db[self.mongo_collection] # 创建唯一索引确保product_id不重复 self.collection.create_index([(product_id, pymongo.ASCENDING)], uniqueTrue)有了唯一索引后Pipeline里就不能直接用insert_one了否则遇到重复键会抛DuplicateKeyError。改为update_one配合$set操作符实现“有就更新没有就插入”def process_item(self, item, spider): data dict(item) self.collection.update_one( {product_id: data[product_id]}, {$set: data}, upsertTrue ) return item这段代码是增量爬虫的核心。跑定时任务的时候新数据会插入已有数据会更新不会产生重复也不会覆盖掉新增的字段。提示如果数据量特别大几千万级别的集合建唯一索引的时间会很长而且会阻塞写入。建议在建索引时加上backgroundTrue参数让索引在后台构建不影响前端写入。3.3 批量写入与性能优化一次插入一条数据在数据量小的时候没感觉但爬虫跑到高峰每秒几十条请求insert_one的性能就会成为瓶颈。我实测过百万级别数据用单条插入和批量插入耗时差距在5到10倍。采用批量写入的Pipelineimport pymongo from itemadapter import ItemAdapter class MongoBulkPipeline: BATCH_SIZE 500 MAX_BUFFER_TIME 10 def __init__(self, mongo_uri, mongo_db, mongo_collection): self.mongo_uri mongo_uri self.mongo_db mongo_db self.mongo_collection mongo_collection self.buffer [] self.last_flush_time 0 classmethod def from_crawler(cls, crawler): return cls( mongo_uricrawler.settings.get(MONGO_URI, mongodb://localhost:27017), mongo_dbcrawler.settings.get(MONGO_DB, scrapy_data), mongo_collectioncrawler.settings.get(MONGO_COLLECTION, items) ) def open_spider(self, spider): import time self.client pymongo.MongoClient(self.mongo_uri, maxPoolSize50) self.db self.client[self.mongo_db] self.collection self.db[self.mongo_collection] self.last_flush_time time.time() def close_spider(self, spider): self.flush() self.client.close() def process_item(self, item, spider): self.buffer.append(dict(item)) if len(self.buffer) self.BATCH_SIZE: self.flush() return item def flush(self): import time if not self.buffer: return try: self.collection.insert_many(self.buffer, orderedFalse) except pymongo.errors.DuplicateKeyError: for doc in self.buffer: try: self.collection.insert_one(doc) except pymongo.errors.DuplicateKeyError: pass self.buffer [] self.last_flush_time time.time()这里面有两个关键参数BATCH_SIZE和MAX_BUFFER_TIME。BATCH_SIZE控制内存里积攒多少条再写入设置太小起不到批量效果设置太大会占用太多内存尤其是Item里带大文本或图片URL列表的时候500是一个比较平衡的值。MAX_BUFFER_TIME是防止爬虫流量太低导致Item一直攒不满数据迟迟不落库这里只是演示了思路实际代码里需要在每条Item进来时判断是否超时。批量写入的orderedFalse参数很关键它告诉MongoDB这批文档即使中间有一条失败了其他的不用管继续写入。这在大批量场景下性能优势非常明显。3.4 字段清洗与类型转换爬虫抓回来的数据上面还有一层脏价格是字符串¥29.90时间格式五花八门有的字段缺失有的嵌套层级不一致。全部丢给MongoDB存下来后面做分析和展示的时候就会很痛苦。我的习惯是分层处理简单的清洗逻辑就在Item里用Loader统一处理复杂的清洗逻辑才放Pipeline。比如价格字段的清洗先用Scrapy的Loader做一次预处理import re from scrapy.loader import ItemLoader from itemadapter import ItemAdapter def clean_price(value): if isinstance(value, str): # 移除货币符号和空格保留数字 match re.search(r(\d\.?\d*), value.replace(,, )) return float(match.group(1)) if match else 0.0 return float(value)Pipeline里针对特殊字段做二次清洗from itemadapter import ItemAdapter class CleanPipeline: def process_item(self, item, spider): adapter ItemAdapter(item) # 统一时间字段为ISO格式 if adapter.get(crawl_time): from datetime import datetime, timezone dt adapter[crawl_time] if isinstance(dt, str): adapter[crawl_time] datetime.fromisoformat(dt).astimezone(timezone.utc) # 处理缺失字段给一个默认值 for field in adapter.field_names(): if adapter.get(field) is None: adapter[field] if field ! price else 0.0 return item字段清洗这一步直接决定了后面数据用起来顺不顺手。我在项目里见过太多“存进去是垃圾查出来还是垃圾”的案例。爬虫代码能跑通不算本事数据入库后能直接用于统计、反查、报表才算真正落地。4. Settings配置与Pipeline启用的正确姿势4.1 Pipeline优先级与配置项写好Pipeline后要在settings.py里启用ITEM_PIPELINES { myproject.pipelines.CleanPipeline: 300, myproject.pipelines.MongoBulkPipeline: 800, }数字越小优先级越高越先执行。所以清洗权重设300Mongo存储设800意思是先清洗干净再入库。这个顺序很重要如果反过来脏数据就直接存库里了。另外在settings里加上MongoDB的连接配置MONGO_URI mongodb://root:your_passwordlocalhost:27017 MONGO_DB scrapy_data MONGO_COLLECTION items连接字符串里如果启用了认证格式是mongodb://用户名:密码主机:端口/。我一开始习惯把用户名密码直接写死在Pipeline代码里后来发现不同的环境开发、测试、生产需要连不同的库每次都要改代码。改成从settings读取后部署时只需要改配置文件代码完全不用动。4.2 多Pipeline协同的调度逻辑实际项目里我通常启用三个Pipeline协同完成数据入库优先级Pipeline职责300CleanPipeline字段清洗、类型转换500DuplicatePipeline重复数据过滤基于Redis Set800MongoBulkPipeline批量写入MongoDB有个细节要注意process_item返回的Item会被传给下一个Pipeline如果某个Pipeline返回了DropItem这个Item就会被丢弃后面的Pipeline不会执行。所以重复过滤要在清洗之后、入库之前执行这样即使丢弃了也不会影响数据质量。我在DuplicatePipeline里用的是Redis的Set做去重因为Redis的SADD命令性能极高而且天然线程安全。对product_id做MD5后存入Set如果返回0说明已存在直接丢import redis import hashlib from scrapy.exceptions import DropItem class DuplicatePipeline: def __init__(self, redis_url): self.redis redis.Redis.from_url(redis_url) self.seen_key crawler:seen:product_id classmethod def from_crawler(cls, crawler): return cls(crawler.settings.get(REDIS_URL, redis://localhost:6379/0)) def process_item(self, item, spider): fp hashlib.md5(item[product_id].encode()).hexdigest() if self.redis.sadd(self.seen_key, fp): return item raise DropItem(fDuplicate item found: {item[product_id]})这样做的优势是做分布式爬虫时多个节点共享同一个Redis天然全局去重。后面章节会专门讲。5. 分布式爬虫中的MongoDB与Redis协同5.1 方案选型Scrapy-Redis的经典架构单机Scrapy爬虫跑业务总有资源上限。CPU、内存、带宽都有瓶颈更麻烦的是单点故障——爬虫跑了两天服务器重启了一下任务全丢了得从头再来。分布式爬虫一般用Scrapy-Redis方案思路是把请求队列从Scrapy默认的内存Queue换成Redis里的队列多个节点共享同一个请求池谁空闲谁去取任务互不冲突。这样既解决了任务共享和去重问题又实现了天然负载均衡。架构里的角色分工是Redis负责调度和去重MongoDB负责存储最终数据。Scrapy-Redis这个库继承并重写了Scheduler和DupeFilter把原来在Scrapy进程内部的东西搬到了Redis上。我用的是 kencx/scrapy-redis 维护的版本兼容性更好一些。5.2 配置分布式Spider要点用Scrapy-Redis写分布式爬虫爬虫的基类从scrapy.Spider换成RedisSpider逻辑上最大的变化是起始URL不再写在start_urls里而是从Redis的一个list里读from scrapy_redis.spiders import RedisSpider class ProductSpider(RedisSpider): name product redis_key product:start_urls def parse(self, response): # 解析逻辑同普通Spider ...启动爬虫后向Redis里推送起始URLredis-cli lpush product:start_urls https://example.com/products/1所有跑这个爬虫的节点都会从product:start_urls这个list里取URL来爬。每次请求的URL会经过Redis的DupeFilter去重保证同一时刻不会有两个节点同时爬同一个页面。数据存储环节完全复用第四节的MongoDB Pipeline因为PyMongo的连接配置天然支持线程池多台机器同时写入同一个MongoDB集合完全没有问题。唯一要做的就是给集合加好索引别在高峰期写入的时候建索引否则会堵住写请求。5.3 爬虫状态与断点续爬分布式还有一个大问题任务中断。某个节点挂了、网络断了跑到一半的任务怎么恢复Scrapy-Redis提供的机制是每个请求的状态都存在Redis里只要任务队列还在重新启动爬虫就能接着跑不用重头爬。不过MongoDB存的数据也存在这种问题如果写了一半进程崩溃这批数据就丢了。所以我上生产环境做分布式爬虫时MongoDB集群必开副本集至少一主一从加仲裁虽然写性能会有一点损耗但换来了数据安全性。爬虫数据丢了不是小事跑了一整天的数据一夜之间没了你会非常被动。6. MongoDB查询与聚合存进去之后的正确打开方式6.1 常用查询语法速查数据存进去了最常用的就是查询。MongoDB用Python的PyMongo查询语法和你熟悉的Mongo Shell一模一样下面是我项目里最高频的几个查询方式# 连接数据库 import pymongo client pymongo.MongoClient(mongodb://localhost:27017) db client[scrapy_data] col db[items] # 等值查询 result col.find({category: 手机}) # 比较查询价格大于3000 result col.find({price: {$gt: 3000}}) # 多条件查询 result col.find({category: 手机, price: {$lt: 5000}}) # 嵌套字段查询针对specs里的内存信息 result col.find({specs.memory: 12GB}) # 数组字段查询images列表有3个元素 result col.find({images.2: {$exists: True}}) # 排序和分页 result col.find().sort(price, pymongo.DESCENDING).limit(20).skip(40) # 只返回需要的字段 result col.find({}, {name: 1, price: 1, _id: 0})有几个容易踩的坑值得说一下。第一find()返回的是游标而不是列表不要直接对它做下标操作要么先转list要么limit和skip配合分页遍历。第二排序字段一定要建索引否则数据量到百万级别后排序会非常慢甚至直接把MongoDB的CPU打满。第三嵌套字段查询的关键是双引号包裹整个字段路径写成specs.memory这个点号在MongoDB里是路径分隔符不能省略。6.2 聚合函数实战统计爬虫数据最常做的操作就是统计比如按类别算商品平均价格、统计每天抓取的商品数量、找出价格最高的前10个商品。这种聚合统计在MongoDB里用聚合管道非常方便语法比写Python脚本在内存里处理快几个数量级。统计每个类别的商品数量、平均价格、最高价格pipeline [ {$group: { _id: $category, count: {$sum: 1}, avg_price: {$avg: $price}, max_price: {$max: $price} }}, {$sort: {count: -1}}, {$limit: 20} ] results col.aggregate(pipeline) for r in results: print(r)统计每天抓取的商品数量pipeline [ {$group: { _id: {$dateToString: {format: %Y-%m-%d, date: $crawl_time}}, count: {$sum: 1} }}, {$sort: {_id: 1}} ]这里有一个在使用$group时特别容易犯的错误_id字段不写或者写错会导致所有文档被分到同一组统计结果就全错了。$group的_id指定分组的依据其他字段用聚合操作符计算写完建议先用小样本数据验证结果是否正确。提示聚合管道在数据量大时建议分阶段加$match先过滤再$group这样能大幅减少内存消耗。另外如果频繁做固定模式的统计可以开MongoDB的增量视图或者定时跑聚合后把结果写入新集合不要每次都全量聚合。6.3 数据同步到其他系统的通道MongoDB在爬虫架构里通常作为原始数据层后续要把数据同步到Elasticsearch做全文检索、同步到ClickHouse做OLAP分析或者同步到MySQL给业务系统用都很常见。最简单的同步方式是写定时脚本find出增量数据再批量写出去。但增量怎么判断我习惯在Items里加一个crawl_time时间戳配合MongoDB的_id自带时间戳特性同步脚本用_id的ObjectId.getTimestamp()就能完成增量识别不需要额外存一个冗余字段。from bson import ObjectId import datetime # 获取最近24小时的新数据 cutoff ObjectId.from_datetime(datetime.datetime.utcnow() - datetime.timedelta(hours24)) result col.find({_id: {$gte: cutoff}})这个技巧很好用因为_id天然带时间信息不占额外存储空间而且按_id范围查还能走索引。7. 常见问题与排查技巧实录7.1 认证失败或连接超时在写连接串的时候密码里如果包含、:、/这些特殊字符会导致连接解析失败。必须先做URL编码Python里有现成的方法from urllib.parse import quote_plus username quote_plus(admin) password quote_plus(password) uri fmongodb://{username}:{password}localhost:27017还有个常见情况是MongoDB启动了认证但连接串没带用户名密码或者用户名密码正确但认证数据库指定错误。注意创建用户时的db很重要连接串里的/后面的部分对应认证库默认是admin。如果用户建在了特定数据库下比如scrapy_data连接串要写成mongodb://root:passlocalhost:27017/scrapy_data。7.2 编码问题Unicode和UTF-8爬取的数据来自各种网站编码五花八门。Scrapy的response默认按Content-Type头里的charset解码有的网站头信息是错的或者干脆不返回charset解码出来就是乱码。在Spider里拿到response后先自己检查一下# 强制重新编码 import chardet def parse(self, response): if response.encoding ! utf-8: text response.body detected chardet.detect(text) response response.replace(bodytext.decode(detected[encoding]).encode(utf-8))乱码数据一旦写进MongoDB想再清洗就得写复杂的正则和替换逻辑所以源头解决最省事。PyMongo连接串也可以指定编码client pymongo.MongoClient(uri, wmajority, journalTrue)w参数是写关注级别majority表示要多数副本节点确认才算写成功对数据安全要求高的场景可以设置但会降低写入性能爬虫场景一般用默认的w1就够了。7.3 批量插入中断与部分失败批量插入最怕的就是中间有脏数据导致整个批次失败。前面Pipeline代码里我用了insert_many(orderedFalse)就是防止一条坏数据拖垮整个批次。但如果集合里建了唯一索引数据里确实存在重复的keyinsert_many依然会抛异常异常信息里会有WriteError列表可以逐个分析。更稳妥的方案是批量插入前先调用delete_many清理目标集合里的脏数据或者先把数据写入一个临时集合等验证完毕再通过聚合管道合并到正式集合。我一般用后者做数据质量保障。7.4 连接池耗尽导致写入卡死Scrapy是异步框架爬虫的并发很高如果Pipeline里每个连接都用一个MongoDB连接且不释放连接池很快就会耗尽写入请求全部阻塞。PyMongo默认有连接池MongoClient默认的maxPoolSize是100如果并发太高可以调大client pymongo.MongoClient(uri, maxPoolSize200, minPoolSize50)注意连接池大小要参考Scrapy的CONCURRENT_REQUESTS配置爬虫并发请求数是100PoolSize最好设为它的两倍以上避免连接竞争。如果设置了CONCURRENT_REQUESTS32那PoolSize给个100绰绰有余。7.5 索引失效导致查询慢MongoDB里如果查询条件里的字段没有建索引数据量大了以后。db.items.createIndex({category: 1, price: -1})组合索引的顺序有讲究等值条件的字段放前面排序或范围条件的字段放后面。你经常用category和price组合查询那就建{category:1, price:-1}1和-1表示升序和降序会影响排序方向如果查询里是按价格降序排别建反了。注意索引不是越多越好每个索引在写入时都要额外维护。爬虫场景写入频率高索引数量太多会拖慢写入性能。只对高频查询和排序字段加索引其他字段一律裸奔。8. 实战经验的一些补充8.1 从一个实际项目看完整链路之前做一个电商平台的价格监控项目架构就是Scrapy加MongoDB。具体流程是这样爬虫每半小时跑一次从商品列表页抓取商品ID、名称、价格、评论数、上架日期进入详情页抓取规格参数、SKU列表、商家信息全部解析成Item。Pipeline里先清洗价格和日期字段然后用Redis Set对商品ID去重最后批量写入MongoDB。这个项目跑了三个月MongoDB里累计了一千多万条商品记录。每次需要分析价格波动时写一条聚合管道查询商品的历史价格数组把数据拉出来画趋势图。换成MySQL这套流程会痛苦很多光是表结构设计就要变好几轮。8.2 数据安全与备份策略MongoDB做爬虫存储数据安全很容易被忽视。爬虫数据虽然不像资金数据那么敏感但作为业务分析的基础丢了也非常麻烦。我的备份策略比较简单粗暴每天凌晨用mongodump做一次全量备份保存近7天的备份文件同时开启MongoDB的副本集主节点挂了自动切换。mongodump的具体用法mongodump --urimongodb://root:passwordlocalhost:27017 --dbscrapy_data --out/backup/mongodb/$(date %Y%m%d)还记得第一节课说的MongoDB认证问题吗千万不要把root密码用弱口令MongoDB被勒索的案例比比皆是部署在公网的MongoDB尤其危险务必要开启认证加IP白名单双重防护最好再加一层防火墙限制27017端口只对指定IP开放。8.3 Scrapy与MongoDB版本兼容的一些说明Scrapy框架迭代频繁PyMongo的兼容性整体不错。我常用的是Scrapy 2.11加PyMongo 4.5这两个版本组合跑得很稳。PyMongo 4.x版本的API和旧版有一个变化insert_one和insert_many默认返回InsertOneResult和InsertManyResult对象如果你写了result.inserted_id的代码在4.x里依然能用但如果你用了PyMongo的旧方法比如insert()那是PyMongo 2.x时代的写法3.x就废弃了4.x里已经移除务必用insert_one或insert_many。Scrapy最近的版本里Item和ItemAdapter的API有调整Pipeline里操作Item时最好用itemadapter库的ItemAdapter类它统一了Item和dict的操作方式可以避免不同版本间的兼容性问题。8.4 后续可继续扩展的方向Scrapy加MongoDB这套组合搭好后往上走的路径其实很清晰。并行爬虫可以用Scrapy-Redis做分布式改造存储层可以用MongoDB的分片集群来扩展数据容量。查询层可以接入Elasticsearch把MongoDB作为原始数据层、ES作为检索层各自干最擅长的事。如果是实时性要求高的场景可以加Kafka做消息队列爬虫把Item打进Kafka下游服务再从Kafka消费写入MongoDB实现爬虫和数据入库的完全解耦。不过这个架构对多数项目来说偏重只有数据量真正大到要拆服务的阶段才值得做。我自己在实际项目里还会用MongoDB的Change Streams功能监听集合并实时推送数据变化触发下游分析任务。这个机制配合Scrapy爬虫可以实现数据采集完就立刻触发计算的链路很实用。踩过几次坑之后我现在接手任何爬虫项目存储层几乎是默认选型MongoDB除非有强事务需求才考虑关系型数据库。我个人的体会是Scrapy负责把网页变成结构化数据MongoDB负责把这个数据结构原汁原味地存下来再用Pipeline把两者无缝对接这一套组合上手快、扩展方便、性能也够用非常适合中小团队快速搭建数据采集系统。希望这篇文章里那些踩坑过程和优化思路能帮你少走一些弯路。