我们总说 MongoDB 写数据很简单不就是insertOne和insertMany两行代码的事嘛。但真到了生产环境面对 20 万条真实业务数据你会发现事情远不止“能写进去”这么简单——写入慢、内存涨、主键冲突、网络中断导致数据对不上各种问题全冒出来了。这篇文章就是我最近一次把一批爬虫商品数据大概 20 万条 JSON 文档批量写入 MongoDB 的完整复盘。从最基本的插入方法讲起到批量优化策略、断点续传设计、异常排查再到最终的压测结果整个过程踩了不少坑也沉淀了一些常规文档里不会写的经验分享出来给你参考。如果你正在学 MongoDB或者正准备把一批历史数据导进 MongoDB这篇内容可以直接当“操作手册”用。我会把每一步的设计思路、参数选择和后续排查讲透不光是告诉你“怎么做”更会解释“为什么这么做”。1. 插入前先想清楚数据长什么样写入目标是什么1.1 先盘清源数据再谈插入方案这次要写入的数据是爬虫跑了一周攒下来的商品信息累计 20 万条左右按照日期分成了多个 JSON 文件。单条数据的结构大概是这样的{ product_id: SPU100230001, name: 某品牌无线机械键盘 87 键, category: 数码/外设/键盘, price: 329.00, stock: 156, shop_name: 某某数码专营店, tags: [办公, 机械键盘, 无线], updated_at: 2024-05-12 14:30:22 }字段不多但有几个点需要注意product_id在业务上是唯一的适合作为_id或唯一索引字段。updated_at是字符串格式因为数据源头是 Python 爬虫直接生成的没有做时间类型转换。嵌套字段只有一个tags数组结构相对简单没有深层次嵌套文档。在动手写任何插入代码之前先做两件事第一检查 JSON 文件是否完整有没有截断或空文件第二统计每个文件的数据量方便后面估算批次大小和进度。1.2 明确写入需求单条插入不够必须批量这次需求有三个硬性指标效率优先20 万条数据不能用 for 循环一条一条 insert那是灾难。可断点续传万一写入中途报错退出下次重新运行程序时不能从头再来必须能接着上次的位置继续。数据可排查写入失败时要能快速定位是哪一批、哪一条出了问题而不是在日志里大海捞针。围绕这三点我一开始就确定用insertMany配合自定义批量大小来做同时设计一个简单的断点记录文件来支持续传。方案听起来简单但里面有不少细节要注意下面一章详细拆解。2. 写入方式选型insertOne、insertMany 还是 bulkWrite2.1 从 API 差异看选型逻辑MongoDB 插入文档的 API 主要有三个很多新手分不清区别我帮你梳理一遍API参数形式单次可操作文档数适用场景备注insertOne单个文档1新增一条数据最简单适合低频单条写入insertMany文档数组多批量导入按顺序插入遇到错误默认中止或继续取决于配置bulkWrite操作数组多批量插入/更新/删除混合最灵活可精确控制每一条操作如果你的需求只是“写一批新数据”insertMany就够了。bulkWrite适合数据可能重复、需要同时做“插入”和“更新”的场景比如做数据同步时存在就更新、不存在就插入。这次是纯新增商品数据所以我直接用insertMany。2.2 为什么不用 for 循环 insertOne拿 20 万条数据来算如果每条insertOne一次就算网络延迟只有 2ms光网络往返就要 400 秒再加上 MongoDB 服务端处理时间基本要 10 分钟以上。而使用insertMany分批发每批 500~1000 条总共只需要 200~400 次网络往返速度快一到两个数量级。2.3 要不要开 ordered 参数insertMany默认ordered: true意思是按数组顺序逐条写入遇到错误立即返回并且之前的写入保留后面的数据不再执行。这个参数默认值“安全”但性能不是最优。如果你希望“尽量多写入、失败的不阻塞后续”可以设置ordered: false。这样 MongoDB 会并行处理写入遇到错误会继续尝试后续文档最后统一返回错误信息。我这次设置的是ordered: false原因有两个这批数据每一行都是独立商品相互之间没有依赖独立写入即使某条失败也不影响其他商品继续入库。配合功能这样能最大化吞吐。注意insertMany单次插入的文档总大小有限制默认上限是 48MB16MB 是单文档上限48MB 是从 MongoDB 3.6 开始对批量写入的限制。如果一批数据超过这个限制需要调小批次大小。3. 核心实操分批写入完整流程3.1 确定合理的批次大小批次大小的选择直接影响写入性能。批太小网络往返次数多吞吐上不去批太大单次占用的内存和耗时都高而且一旦出错重试代价大。我实测下来500 到 1000 条一批是比较稳的区间。这次我选的是 500原因如下单条商品 JSON 换算成 BSON 后平均大约 300~500 字节500 条也就是 150~250KB远低于 48MB 的限制。网络传输和 MongoDB 写入在 500 条这个量级上能达到较好平衡。出错时重试成本可控而且日志里能按批次清晰定位。如果你的单条文档很大例如有几 MB 的图片 Base64那批次大小要主动下调建议按“总字节数”来估算而不是只看条数。一般来说控制单批总大小在 10MB 以内比较稳妥。3.2 带断点续传的 Python 实现这次我用的语言是 Python驱动是pymongo。完整实现可以拆成三部分读取文件、分批插入、记录断点。import json import os from pymongo import MongoClient, errors MONGO_URI mongodb://localhost:27017 DB_NAME shop COLLECTION_NAME products DATA_DIR ./data BATCH_SIZE 500 CHECKPOINT_FILE ./checkpoint.json def get_collection(): client MongoClient(MONGO_URI, serverSelectionTimeoutMS5000) db client[DB_NAME] return db[COLLECTION_NAME] def load_checkpoint(): if os.path.exists(CHECKPOINT_FILE): with open(CHECKPOINT_FILE, r, encodingutf-8) as f: return json.load(f) return {} def save_checkpoint(checkpoint): with open(CHECKPOINT_FILE, w, encodingutf-8) as f: json.dump(checkpoint, f, ensure_asciiFalse, indent2) def process_file(file_path, collection, checkpoint): file_name os.path.basename(file_path) # 已经处理过的文件直接跳过 if checkpoint.get(file_name) done: print(f跳过已完成的文件: {file_name}) return # 读取当前文件的断点默认从 0 开始 next_offset checkpoint.get(file_name, 0) with open(file_path, r, encodingutf-8) as f: data json.load(f) total len(data) print(f文件 {file_name} 共 {total} 条数据从第 {next_offset} 条继续) current next_offset while current total: batch data[current:current BATCH_SIZE] try: collection.insert_many(batch, orderedFalse) except errors.BulkWriteError as e: # 批量写入错误打印错误详情 print(f批量写入出现错误发生在 offset{current}, 文件{file_name}) for error in e.details.get(writeErrors, []): print(f错误索引: {error.get(index)}, 错误信息: {error.get(errmsg)}) # 无论有没有错误都把断点推进到本次批次的末尾 current BATCH_SIZE checkpoint[file_name] current save_checkpoint(checkpoint) print(f进度: {current}/{total}) checkpoint[file_name] done save_checkpoint(checkpoint) print(f完成文件: {file_name}) def main(): collection get_collection() checkpoint load_checkpoint() files [f for f in os.listdir(DATA_DIR) if f.endswith(.json)] for file_name in files: file_path os.path.join(DATA_DIR, file_name) process_file(file_path, collection, checkpoint) print(全部文件处理完毕) if __name__ __main__: main()这段代码有几个设计要点值得展开说第一断点粒度是按“文件 偏移量”记的所以程序中途崩了重启后会从最后一批的位置继续写。注意这里的“继续写”不是“跳过这一批”而是“从这一批的开头重试”。所以如果你担心批量中有部分成功部分失败导致重复可以在插入前先用product_id去重或者给_id设成product_id用幂等写入兜底。第二save_checkpoint在每批处理完就调用一次文件里存的是 int 类型的 offset。这种做法比“全部完成后统一保存”靠谱得多能承受任意时刻的进程杀掉和断电。第三异常处理这里我用的except errors.BulkWriteError并且只打印错误详情没有中断整个流程。这样的选择是有意的商品数据量大个别字段格式有问题并不影响整体入库先记录后修复比一遇到脏数据就卡死整个任务更合理。3.3 关于_id的去重设计MongoDB 默认会对每一条文档生成_idObjectId但在导入业务数据时我更建议把_id显式指定为业务主键。这样做有几个好处重复插入时MongoDB 会直接报主键冲突天然防重。后续做增量更新时可以配合replaceOne或updateOne做幂等写入。查询时如果用product_id查走_id索引是最快的。我在插入前做了一步数据清洗在每一条商品数据里补上_id字段赋值为product_id的值。如果你的场景里没有天然唯一键也可以考虑在updated_at或其他业务字段上建唯一索引效果类似。# 插入前的数据预处理 for item in batch: if _id not in item: item[_id] item.get(product_id)3.4 Windows 上装 MongoDB 的额外提醒这次是在 Windows 环境下开发的。热词里频繁出现“windows 上装 mongodb”“mongodb安装失败”说明在 Windows 上把 MongoDB 跑起来确实坑不少。我只补充一点最关键的MongoDB 在 Windows 上默认不会注册成系统服务手动启动或开机自启都很别扭。如果你的机器还没装好 MongoDB最简单的流程是官网下载 MongoDB Community Server 的 zip 包解压到C:\mongodb然后在C:\mongodb\data建好数据目录用管理员权限打开终端执行C:\mongodb\bin\mongod.exe --dbpath C:\mongodb\data想注册成 Windows 服务可以这样C:\mongodb\bin\mongod.exe --dbpath C:\mongodb\data --logpath C:\mongodb\log\mongod.log --install如果安装失败八成是目录权限、杀毒软件拦截或者 27017 端口被占用。优先查看日志文件mongod.log比在网上盲搜关键词管用。4. 实操过程中的性能数据与观察4.1 实测500 条一批20 万条数据总耗时多少我用本地一台配置很普通的电脑8 核 i5、16GB 内存、SSD做了完整写入测试。MongoDB 版本 4.4.30单机没有任何副本集和分片配置。测试结果批次大小总耗时MongoDB CPU 占用峰值备注100约 42 秒约 30%网络往返较多速度一般500约 18 秒约 55%综合表现最好1000约 16 秒约 65%速度略快但内存占用更高2000约 18 秒约 80%开始出现明显的写锁竞争从这个数据能看出批次从 100 提升到 500速度提升非常明显从 500 提升到 1000增益已经很小。批次太大反而会让 MongoDB 服务端的写请求排队CPU 打满最终耗时也没有明显下降。所以如果你的数据在几十万条这个量级批次大小首选 500不用犹豫。4.2 观察 MongoDB Compass 中的数据变化一边跑脚本一边打开 MongoDB Compass 看数据条数变化是很直观的验证手段。Compass 的 Collection 标签页会显示当前 collection 的文档总数写入过程中能实时看到数字跳动。这里有个小细节Compass 的文档计数不是实时的会有一点延迟默认大概 1~2 秒刷新一次。如果你发现数字不动先看一眼是不是页面停在没有自动刷新的状态不要误以为程序卡住了。4.3 为什么 MongoDB 写入这么快理解底层机制很多人第一次用 MongoDB 批量写数据都会被它的速度惊到。这套速度背后其实是几个设计共同作用的结果内存映射文件MongoDB 使用 WiredTiger 存储引擎写入时会先进入内存再异步刷盘所以单次写入的响应非常快。批量写入合并insertMany在驱动层面就会把多文档请求合并减少了网络和协议处理的开销。默认非强制 fsyncMongoDB 默认写关注是w:1意味着主节点收到并写入内存就算成功不等待所有副本单机部署时没有副本等待更少。理解了这几层你就明白了为什么“先攒批再插”远比“来一条插一条”高效。5. 常见问题速查表与避坑经验5.1 高频问题清单我把自己踩过、以及身边同事常遇到的问题整理成一张表方便你排查时直接对照问题现象可能原因解决方案insertMany报错document too large单条 BSON 超过 16MB检查文档里是否嵌入了大文件或 base64 字符串批量插入报E11000 duplicate key error_id或唯一索引冲突显式指定_id或用updateOne(upsertTrue)做幂等写入插入速度快但 CPU 很高单批次太大写锁竞争严重调小BATCH_SIZE到 500 左右45 秒后连接超时网络不通或serverSelectionTimeoutMS太短在 MongoClient 里调大超时时间检查防火墙插入完成后查不到数据用了事务没提交或写入到了别的库检查事务提交逻辑确认库名、集合名和客户端连接Windows 下 MongoDB 启动后马上退dbpath不存在或权限不足提前mkdir数据目录确认目录有读写权限看日志5.2 断点续传与重复数据如何共存有朋友看完上面的代码问断点续传如果是从“当前批次开头”重新写那之前批次里成功插入的数据会不会重复插入答案是如果 MongoDB 端有唯一索引比如_id就是product_id重复插入会直接报主键冲突但不会中断其他文档的写入而且也不会产生脏数据。所以断点续传的可靠性很大程度上依赖你_id设计得是否合理。这也是我前面强调“把_id设成业务主键”的原因——它不只是为了查询快更是为了导入数据时可重试、可对账。5.3 插入前一定要做“数据体检”一次导入几万条数据前别急着写代码。先写个小脚本做数据体检检查这几个维度必填字段是否存在比如product_id、name。字段类型是否统一比如price是数字还是字符串。JSON 文件是否完整在 Python 里能不能正常json.load。是否有重复的product_id。如果这些基础问题没提前筛掉后面写入时你会被一条条脏数据反复打断调试成本极高。6. 更进阶的玩法upsert 写入与双写对账6.1 用bulkWrite实现存在即更新不存在即插入第一批数据入库后后续爬虫还会增量爬取这时候再遇到已经存在的商品就不能直接插入了会报主键冲突。bulkWrite配合 upsert 就很适合这种场景。from pymongo import UpdateOne requests [] for item in batch: requests.append( UpdateOne( {_id: item[_id]}, {$set: item}, upsertTrue ) ) collection.bulk_write(requests, orderedFalse)这行代码的效果是文档存在就更新对应字段不存在就插入整条。清洗脚本和同步脚本共用这套逻辑后面再做增量更新时基本不需要改代码。6.2 写完以后怎么确认数据没丢导入完 20 万条数据光看“程序没有报错”是不够的。我习惯再做一道 “双写对账”把源 JSON 里的行数、字段求和值跟 MongoDB 里查出来的总数、求和值逐一比对。如果两边对得上这次导入才算真正收工。total_in_mongo collection.count_documents({}) print(fMongoDB 文档总数: {total_in_mongo})如果数量对不上就从断点日志和错误日志里逐批排查缩小问题范围。这种对账成本很低但能避免后面业务使用脏数据时才发现问题算是性价比极高的一个步骤。7. 最后分享一点实际操作心得整套流程走完我最想强调的是MongoDB 插入文档这件事心智负担可以很小也可以很大差别就在于你有没有在“写之前”做足功课。数据清洗做得好、_id设计合理、批次大小恰到好处、断点记录到位后面所有环节都会顺。如果只记住一句话那就是**别用循环单条 insertOne一定要分批 insertMany每条数据最好带上业务主键做_id一定要有断点续传和对账机制。**这三点做到位哪怕数据量再翻几倍也只改批次大小和机器配置的问题不用改架构。我其实是建议每个人第一次大批量导入前都先用 1000 条数据跑个测试监控一下 MongoDB 的 CPU、内存和耗时再决定批次大小。因为不同机器、不同数据大小、不同网络环境最优批次参数真的差很多。拿着别人的“标准答案”直接用不如自己测一遍来得踏实。