如何为MongoDB中的子文档批量补添_id字段？

针对10万条文档的规模，推荐两种高效的迁移方案，避免一次性加载全量数据导致内存压力：

方法一：MongoDB Shell 脚本（直接数据库端操作） #

适合快速执行，无需额外开发环境，优先推荐用聚合管道批量更新（MongoDB 4.2+支持），效率远高于客户端遍历：

高效批量更新（聚合管道方式） #

// 替换为你的集合名称和子文档数组字段名
const COLLECTION = "your_collection";
const SUBDOCS_FIELD = "subdocs_array";

db.getCollection(COLLECTION).updateMany(
  // 匹配存在子文档数组的文档
  { [SUBDOCS_FIELD]: { $exists: true, $type: "array" } },
  [
    {
      $set: {
        [SUBDOCS_FIELD]: {
          $map: {
            input: `$${SUBDOCS_FIELD}`,
            as: "subdoc",
            in: {
              // 仅为无_id的子文档生成新ObjectId，已有_id的保持不变
              $mergeObjects: [
                "$$subdoc",
                { _id: { $cond: [{ $not: ["$$subdoc._id"] }, ObjectId(), "$$subdoc._id"] } }
              ]
            }
          }
        }
      }
    }
  ]
);

print(`完成更新：为符合条件的子文档批量生成ObjectId`);

逐文档迭代方式（兼容旧版MongoDB） #

如果你的MongoDB版本低于4.2，用游标遍历避免内存溢出：

const COLLECTION = "your_collection";
const SUBDOCS_FIELD = "subdocs_array";
const BATCH_SIZE = 100; // 每处理100条打印进度，可根据服务器性能调整
let processedCount = 0;

db.getCollection(COLLECTION).find().forEach(doc => {
  if (!doc[SUBDOCS_FIELD] || !Array.isArray(doc[SUBDOCS_FIELD])) return;

  const updatedSubdocs = doc[SUBDOCS_FIELD].map(subdoc => {
    return subdoc._id ? subdoc : { ...subdoc, _id: ObjectId() };
  });

  db.getCollection(COLLECTION).updateOne(
    { _id: doc._id },
    { $set: { [SUBDOCS_FIELD]: updatedSubdocs } }
  );

  processedCount++;
  if (processedCount % BATCH_SIZE === 0) {
    print(`已处理 ${processedCount} 条文档`);
  }
});

print(`迁移完成，共处理 ${processedCount} 条包含子文档数组的文档`);

方法二：Node.js 脚本（适合集成到应用流程） #

如果需要将迁移逻辑嵌入到业务代码或定时任务中，用MongoDB Node.js驱动实现：

安装依赖 #

npm install mongodb

迁移脚本 #

const { MongoClient, ObjectId } = require('mongodb');

async function runMigration() {
  const MONGO_URI = 'mongodb://localhost:27017'; // 替换为你的MongoDB连接串
  const DB_NAME = 'your_database'; // 替换为数据库名
  const COLLECTION_NAME = 'your_collection'; // 替换为集合名
  const SUBDOCS_FIELD = 'subdocs_array'; // 替换为子文档字段名
  const BATCH_SIZE = 100;

  let client;
  try {
    client = await MongoClient.connect(MONGO_URI);
    const db = client.db(DB_NAME);
    const collection = db.collection(COLLECTION_NAME);
    const cursor = collection.find({ [SUBDOCS_FIELD]: { $exists: true, $type: "array" } });

    let processedCount = 0;
    let bulkOps = [];

    while (await cursor.hasNext()) {
      const doc = await cursor.next();
      const updatedSubdocs = doc[SUBDOCS_FIELD].map(subdoc => {
        return subdoc._id ? subdoc : { ...subdoc, _id: new ObjectId() };
      });

      bulkOps.push({
        updateOne: {
          filter: { _id: doc._id },
          update: { $set: { [SUBDOCS_FIELD]: updatedSubdocs } }
        }
      });

      // 批量提交更新
      if (bulkOps.length >= BATCH_SIZE) {
        await collection.bulkWrite(bulkOps);
        processedCount += bulkOps.length;
        console.log(`已批量处理 ${processedCount} 条文档`);
        bulkOps = [];
      }
    }

    // 处理剩余的文档
    if (bulkOps.length > 0) {
      await collection.bulkWrite(bulkOps);
      processedCount += bulkOps.length;
    }

    console.log(`迁移完成，共处理 ${processedCount} 条文档`);
  } catch (err) {
    console.error('迁移失败:', err);
  } finally {
    if (client) await client.close();
  }
}

runMigration();

通用注意事项 #

1. 先备份数据：执行迁移前务必备份目标集合，避免数据丢失：

   // MongoDB Shell中备份
   db.your_collection.copyTo("your_collection_backup");
   

2. 选择合适的时间执行：如果是生产环境，建议在低峰期执行迁移，避免影响业务。
3. 调整批量大小：根据服务器内存和CPU性能，调整BATCH_SIZE参数，平衡速度和资源占用。

内容的提问来源于stack exchange，提问作者Bear Bile Farming is Torture