如何使用MongoDB Aggregate聚合start事件及后续至下一个start前的log事件

需求场景 #

现有MongoDB集合结构如下：

[
  {timestamp: 100, type:'start'},
  {timestamp: 101, type:'log'},
  {timestamp: 102, type:'log'},
  {timestamp: 103, type:'log'},
  {timestamp: 200, type:'start'},
  {timestamp: 201, type:'log'},
  {timestamp: 202, type:'log'}
]

需要通过聚合操作，把每个start事件及其后续所有log事件（直到下一个start前）合并成一组，期望输出：

[
  {
    "events": [
      {timestamp: 100, type:'start'},
      {timestamp: 101, type:'log'},
      {timestamp: 102, type:'log'},
      {timestamp: 103, type:'log'}
    ]
  },
  {
    "events": [
      {timestamp: 200, type:'start'},
      {timestamp: 201, type:'log'},
      {timestamp: 202, type:'log'}
    ]
  }
]

聚合实现方案 #

通过以下4个阶段的聚合管道即可完成需求：

1. 生成分组ID
   按时间戳排序文档，给每个文档标记所属分组ID：遇到type: 'start'事件时分组ID加1，后续log事件继承当前ID，直到下一个start出现。
2. 按分组ID聚合文档
   将同一分组ID的所有文档收集到events数组中。
3. 清理冗余字段
   移除每个文档里的分组ID，只保留需要的timestamp和type字段。
4. 保证输出顺序
   按每组第一个事件的时间戳排序，确保输出顺序和原数据一致。

完整聚合代码 #

db.collection.aggregate([
  {
    $setWindowFields: {
      sortBy: { timestamp: 1 },
      output: {
        group_id: {
          $sum: {
            $cond: [{ $eq: ["$type", "start"] }, 1, 0]
          }
        }
      }
    }
  },
  {
    $group: {
      _id: "$group_id",
      events: { $push: "$$ROOT" }
    }
  },
  {
    $project: {
      _id: 0,
      events: {
        $map: {
          input: "$events",
          as: "item",
          in: { timestamp: "$$item.timestamp", type: "$$item.type" }
        }
      }
    }
  },
  {
    $sort: { "events.0.timestamp": 1 }
  }
])

代码说明 #

• $setWindowFields：确保文档按时间顺序处理，通过条件判断累计分组ID，实现按start事件分割组。
• $group：把同一分组的文档全部推入events数组，完成组内聚合。
• $project：用$map过滤掉每个文档里的group_id字段，只保留业务需要的字段。
• $sort：保证输出的分组按起始事件的时间排序，和原始数据的时序逻辑一致。

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