MongoDB中如何跨集合迁移数据并自定义增减文档字段

针对你需要丢弃旧集合无用字段、保留指定字段同时新增字段的迁移需求，直接按数据量选择对应方案操作即可：

方案1：聚合管道直接迁移（中小数据量首选，操作最简单） #

直接通过MongoDB原生聚合管道，先筛选要保留的字段，再追加新字段，最后直接输出到新集合，全程不用导出导入数据，效率最高。
对应你示例里oldCollection保留k1/k3、丢弃k2、新增k4/k5的场景，直接执行下面的命令：

// 先切换到业务所在的数据库
use 你的目标数据库名

db.oldCollection.aggregate([
  // 第一步：指定要保留的旧字段，不需要的字段直接不写就会被丢弃
  {
    $project: {
      k1: 1,
      k3: 1,
      _id: 1 // _id默认会保留，不需要迁移_id的话把这行改成 _id: 0
    }
  },
  // 第二步：追加新增字段，固定值直接赋值即可，需要动态计算也支持写表达式
  {
    $addFields: {
      k4: "k4对应的固定值",
      k5: "k5对应的固定值"
    }
  },
  // 第三步：写入新集合
  {
    $out: "newCollection"
  }
], { allowDiskUse: true }) // 数据量超过10万条建议开这个参数，避免内存溢出报错

注意：$out会直接覆盖同名新集合，如果你的newCollection已经存在存量数据，把最后一个$out阶段替换成$merge即可实现增量写入，不会覆盖原有数据：

{
  $merge: {
    into: "newCollection",
    on: "_id",
    whenMatched: "merge",
    whenNotMatched: "insert"
  }
}

方案2：批量游标迁移（千万级以上大数据量首选，性能更稳定） #

如果你的旧集合数据量达到百万、千万级别，聚合管道可能出现长时间阻塞、内存占用过高的问题，用批量游标分批读写的方式更稳妥：

use 你的目标数据库名
const BATCH_SIZE = 5000 // 单批处理量，按服务器配置可在1000-10000之间调整
const cursor = db.oldCollection.find({}, { k1: 1, k3: 1 }).batchSize(BATCH_SIZE)
let bulkOp = db.newCollection.initializeUnorderedBulkOp()
let migratedCount = 0

cursor.forEach(doc => {
  // 给每条文档追加新字段
  doc.k4 = "k4对应的取值"
  doc.k5 = "k5对应的取值"
  bulkOp.insert(doc)
  migratedCount++

  // 攒够单批数量就执行写入，重置批量操作对象
  if (migratedCount % BATCH_SIZE === 0) {
    bulkOp.execute()
    bulkOp = db.newCollection.initializeUnorderedBulkOp()
    print(`已完成迁移：${migratedCount} 条`)
  }
})

// 处理最后不足一批的剩余数据
if (migratedCount % BATCH_SIZE !== 0) {
  bulkOp.execute()
}
print(`迁移全部完成，总计迁移 ${migratedCount} 条数据`)

迁移注意事项 #

• 正式迁移前先拿10-100条测试数据跑通流程，校验字段保留、新增逻辑是否符合预期，避免逻辑错误导致全量数据出错
• 生产环境迁移尽量选业务低峰期操作，避免占满实例CPU、内存影响线上业务
• 如果新集合需要建索引，建议等全量数据迁移完成后再创建索引，能大幅提升迁移速度
• 迁移完成后先做两层校验：一是对比新老集合的文档总数是否一致，二是随机抽10-20条文档核对字段、取值是否正确

内容的提问来源于stack exchange，提问作者Prakhar Singh