MongoDB物化视图如何实现输出结果的增量更新？

MongoDB 原生$merge实现的按需物化视图默认只支持全量聚合重刷，没有内置单条变更增量计算能力，要实现「不遍历全量记录、仅更新对应聚合值」的需求，用变更流（Change Streams）+ 定点增量更新的方案即可，性能完全覆盖你单场景2.5万条记录的业务要求。

1. 初始化物化视图结构 #

不要在物化视图里只存孤立的总和值，按你实际的聚合维度（示例中是month、client_id）拆分聚合粒度，单条视图记录结构参考：

{
  "_id": {
    "month": "jan",
    "client_id": "c1"
  },
  "total_salary": 2500,
  "last_updated": ISODate("2024-01-01T00:00:00Z")
}

首次部署时跑一次全量聚合灌入基础数据，后续不需要再执行全量计算：

db.EmployeeSalary.aggregate([
  {
    $group: {
      _id: {
        month: "$month",
        client_id: "$client_id"
      },
      total_salary: { $sum: "$salary" }
    }
  },
  {
    $merge: {
      into: "SalarySummaryMatView",
      whenMatched: "replace",
      whenNotMatched: "insert"
    }
  }
])

2. 配置变更流实现增量更新 #

开启源集合EmployeeSalary的变更流，仅监听数据增、删、改操作，拿到变更前后的文档值后，直接按「旧聚合值 - 旧字段值 + 新字段值」的逻辑定点更新物化视图，全程不需要扫描其他无关记录。

前置配置 #

首先开启源集合的变更前镜像记录能力，才能拿到更新/删除操作发生前的字段旧值：

db.runCommand({
  collMod: "EmployeeSalary",
  changeStreamPreAndPostImages: { enabled: true }
})

注意：变更流需要MongoDB运行在副本集或分片集群模式，单节点测试可以加--replSet参数启动为单节点副本集。

增量更新逻辑 #

针对不同操作类型的增量计算规则：

• 新增记录：直接给对应维度的聚合值加上新记录的salary值，不存在对应维度条目则新建
• 更新/替换记录：计算薪资差值delta = 新salary - 旧salary，直接给对应维度聚合值加delta；如果记录的聚合维度字段（比如月份、所属客户）发生了变更，先从旧维度聚合值中扣减旧薪资，再给新维度聚合值加上新薪资
• 删除记录：直接从对应维度的聚合值中扣减被删除记录的salary值

核心监听代码参考：

// 初始化变更流监听
const changeStream = db.EmployeeSalary.watch(
  [ { $match: { operationType: { $in: ["insert", "update", "replace", "delete"] } } } ],
  {
    fullDocument: "updateLookup",
    fullDocumentBeforeChange: "required"
  }
)

changeStream.on("change", (change) => {
  const session = db.getMongo().startSession()
  session.startTransaction()
  try {
    const view = session.getDatabase(db.getName()).SalarySummaryMatView
    let targetGroup = null
    let delta = 0

    switch(change.operationType) {
      case "insert":
        targetGroup = {
          month: change.fullDocument.month,
          client_id: change.fullDocument.client_id
        }
        delta = change.fullDocument.salary
        break
      case "update":
      case "replace":
        const newDoc = change.fullDocument
        const oldDoc = change.fullDocumentBeforeChange
        const newGroup = { month: newDoc.month, client_id: newDoc.client_id }
        const oldGroup = { month: oldDoc.month, client_id: oldDoc.client_id }
        // 维度未变更，直接算差值
        if (JSON.stringify(newGroup) === JSON.stringify(oldGroup)) {
          targetGroup = newGroup
          delta = newDoc.salary - oldDoc.salary
        } else {
          // 维度变更，先扣旧维度的值
          view.updateOne(
            { _id: oldGroup },
            { $inc: { total_salary: -oldDoc.salary }, $currentDate: { last_updated: true } }
          )
          targetGroup = newGroup
          delta = newDoc.salary
        }
        break
      case "delete":
        const delDoc = change.fullDocumentBeforeChange
        targetGroup = { month: delDoc.month, client_id: delDoc.client_id }
        delta = -delDoc.salary
        break
    }

    // 定点更新聚合值，走主键索引，无全表扫描
    view.updateOne(
      { _id: targetGroup },
      {
        $inc: { total_salary: delta },
        $currentDate: { last_updated: true }
      },
      { upsert: true }
    )

    session.commitTransaction()
  } catch (e) {
    session.abortTransaction()
    throw e
  } finally {
    session.endSession()
  }
})

3. 性能与可靠性说明 #

• 所有物化视图更新都是基于_id（聚合维度复合键）的定点写入，走默认主键索引，单条更新耗时亚毫秒级，哪怕每秒上千次变更也无性能压力
• 变更流自带断点续传能力，监听进程重启后会从上次消费断点继续同步，不会丢数据也不会重复计算
• 2.5万条记录规模下，首次全量初始化耗时仅几十毫秒，增量阶段无任何全量扫描开销

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