MongoDB跨集合数据迁移：嵌入employments数据至person集合并更新employment文档

方案说明 #

使用$merge聚合算子实现，该算子可以将聚合结果直接写入目标集合，支持全量更新，适合大规模数据集处理，不会占用过多客户端内存。 #

第一步：将employments全量数据嵌入person集合 #

在person集合上执行以下聚合操作：

db.person.aggregate([
  // 关联当前person对应的所有employments数据
  {
    $lookup: {
      from: "employments",
      localField: "_id",
      foreignField: "person_id",
      as: "employments"
    }
  },
  // 将结果合并回person集合
  {
    $merge: {
      into: "person",
      on: "_id",
      whenMatched: "merge", // 匹配到文档仅合并新增字段，不覆盖原有其他字段
      whenNotMatched: "discard" // 不存在的文档直接丢弃，不新增无效数据
    }
  }
])

执行完成后person集合会新增employments数组字段，存储对应用户的所有就业数据，和预期结果完全一致。 #

第二步：将person元数据同步到employments集合 #

在employments集合上执行以下聚合操作：

db.employments.aggregate([
  // 关联对应的person元数据
  {
    $lookup: {
      from: "person",
      localField: "person_id",
      foreignField: "_id",
      as: "person_tmp"
    }
  },
  // person_id唯一，把关联得到的数组转为单个对象
  {
    $unwind: "$person_tmp"
  },
  // 新增person字段，保留需要的元数据
  {
    $addFields: {
      person: {
        _id: "$person_tmp._id",
        name: "$person_tmp.name",
        email: "$person_tmp.email",
        contact: "$person_tmp.contact"
      }
    }
  },
  // 移除临时字段和不需要的原有业务字段
  {
    $unset: ["person_tmp", "employment_type", "yearly_income"]
  },
  // 合并回employments集合
  {
    $merge: {
      into: "employments",
      on: "_id",
      whenMatched: "replace", // 直接替换匹配到的文档，确保多余字段被清除
      whenNotMatched: "discard"
    }
  }
])

大规模数据集优化建议 #

• 提前为employments.person_id建立索引，大幅提升$lookup阶段的执行效率
• 可以按_id范围拆分任务分批执行聚合，避免单任务占用过多数据库资源
• 执行操作前先对两个集合做全量备份，避免数据误操作

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