MongoDB合并两个集合并保留时间戳最新文档的实现方法

解决方案 #

你之前想先导入临时集合的思路是完全可行的，完整操作流程如下：

步骤1：导出源集合数据并导入临时集合 #

先从源主机导出目标集合数据：

mongoexport --uri="mongodb://源主机地址/源数据库名" --collection=源集合名 --out=coll_data.json

再将导出的json数据导入到目标主机的临时集合（例如命名为myColl_temp）：

mongoimport --uri="mongodb://目标主机地址/目标数据库名" --collection=myColl_temp --file=coll_data.json

步骤2：执行聚合管道实现按时间戳合并 #

在目标数据库中执行如下聚合管道，即可自动对比同_id文档的timest字段，保留最新版本写入正式集合myColl：

db.myColl_temp.aggregate([
  // 合并临时集合和正式集合的所有文档
  {
    $unionWith: "myColl"
  },
  // 按_id分组，记录同_id下最大的时间戳和所有文档
  {
    $group: {
      _id: "$_id",
      maxTimest: { $max: "$timest" },
      docs: { $push: "$$ROOT" }
    }
  },
  // 筛选出时间戳等于最大值的那个最新文档
  {
    $project: {
      latestDoc: {
        $first: {
          $filter: {
            input: "$docs",
            cond: { $eq: ["$$this.timest", "$maxTimest"] }
          }
        }
      }
    }
  },
  // 将最新文档设为根节点
  {
    $replaceRoot: { newRoot: "$latestDoc" }
  },
  // 合并结果写入正式集合，存在则替换，不存在则插入
  {
    $merge: {
      into: "myColl",
      on: "_id",
      whenMatched: "replace",
      whenNotMatched: "insert"
    }
  }
])

步骤3：清理临时资源 #

合并完成后可删除临时集合myColl_temp和本地导出的json文件，释放存储空间。

Go驱动实现方式 #

如果你要在Go应用中集成这个逻辑，直接调用MongoDB Go驱动的聚合方法执行上述管道即可，示例代码片段如下：

pipeline := mongo.Pipeline{
    bson.D{{Key: "$unionWith", Value: "myColl"}},
    bson.D{{Key: "$group", Value: bson.D{
        {Key: "_id", Value: "$_id"},
        {Key: "maxTimest", Value: bson.D{{Key: "$max", Value: "$timest"}}},
        {Key: "docs", Value: bson.D{{Key: "$push", Value: "$$ROOT"}}},
    }}},
    // 剩余管道节点按上面的JS结构对应转换即可
}
cur, err := tempCollection.Aggregate(context.TODO(), pipeline)
if err != nil {
    // 错误处理
}

注意事项 #

• 导出源集合时建议添加--readConcern majority参数，保证导出的是一致的快照数据，避免导出过程中源集合有修改导致数据异常
• 如果集合数据量较大，可以给timest字段加索引提升聚合效率

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