生产环境MongoDB旧集合批量迁移至新集合的问题解决

核心思路 #

先锁定需迁移的原400万文档的边界，通过范围分页替代skip+limit，配合基于唯一键的幂等写入，确保只处理目标数据集，避免新增文档干扰，同时保证原文档全部迁移完成。

具体步骤 #

1. 锁定迁移数据集边界
   在脚本启动的第一时间，执行查询获取原OldCollection中所有待迁移文档的最大_id（MongoDB的ObjectId自带时间戳，迁移开始后新增的文档_id必然大于该值）：

   const maxMigrationId = await db.collection('OldCollection').find()
     .sort({ _id: -1 })
     .limit(1)
     .toArray()[0]._id;
   

   后续所有读取操作都限定在_id <= maxMigrationId的范围内，彻底排除迁移过程中新增的3000份文档。

2. 用范围分页替代skip+limit
   放弃依赖skip的分页方式，改用_id范围查询实现确定性分批读取：

   • 初始化lastProcessedId为最小ObjectId：ObjectId("000000000000000000000000")
   • 循环执行以下操作，直到返回空结果：

     const batchDocs = await db.collection('OldCollection').find({
       _id: { $gt: lastProcessedId, $lte: maxMigrationId }
     }).sort({ _id: 1 }).limit(1000).toArray(); // 批量大小可根据性能调整
     
     if (batchDocs.length === 0) break;
     
     // 执行文档转换逻辑
     const transformedDocs = batchDocs.map(doc => transform(doc));
     
     // 幂等写入新集合
     const bulkOps = transformedDocs.map(doc => ({
       updateOne: {
         filter: { _id: doc._id },
         update: { $set: doc },
         upsert: true
       }
     }));
     await db.collection('NewCollection').bulkWrite(bulkOps);
     
     // 更新最后处理的ID，作为下一批的起始
     lastProcessedId = batchDocs[batchDocs.length - 1]._id;
     

3. 验证迁移完整性
   迁移完成后，执行两次计数验证：

   • 统计OldCollection中_id <= maxMigrationId的文档数：db.OldCollection.countDocuments({_id: {$lte: maxMigrationId}})
   • 统计NewCollection中的文档数：db.NewCollection.countDocuments()
     若两者数值一致，则说明原400万文档全部迁移完成。

方案优势 #

• 避免新增文档干扰：通过maxMigrationId锁定原数据集，新增文档不会进入读取范围，彻底解决skip+limit的分页偏移问题。
• 幂等写入：基于_id的upsert操作，即使脚本中途重试或重复处理批次，也不会触发11000重复错误，同时确保原文档不会丢失。
• 高效稳定：利用_id的默认索引，范围查询的性能远优于skip+limit，大集合下的分页效率更高。

内容的提问来源于stack exchange，提问作者Daniel Cohen