MongoDB使用dateTrunc分组后按字段汇总对象数组错误计数

我在MongoDB中使用时间序列集合，以1分钟粒度存储各类API的错误计数。API名称未知且可能变更，因此查询最好不要依赖静态API名称。

示例数据集 #

[
  {
    _time: ISODate("2022-03-22T00:00:00.000Z"),
    errors: [
      { api: "shipping", count: 10 },
      { api: "inventory", count: 100 }
    ]
  },
  {
    _time: ISODate("2022-03-22T00:01:00.000Z"),
    errors: [
      { api: "shipping", count: 20 },
      { api: "inventory", count: 200 }
    ]
  },
  {
    _time: ISODate("2022-03-22T00:02:00.000Z"),
    errors: [ { api: "inventory", count: 300 } ]
  },
  {
    _time: ISODate("2022-03-22T00:03:00.000Z"),
    errors: [
      { api: "inventory", count: 400 },
      { api: "account", count: 1 }
    ]
  }
]

已实现的分组代码 #

目前我已经能通过以下代码将文档按2分钟粒度的时间桶分组：

db.collection.aggregate([
  {
    $group: {
      _id: {
        $dateTrunc: {
          date: "$_time",
          unit: "minute",
          binSize: 2
        }
      }
    }
  }
])

需求 #

• 使用$dateTrunc将文档分组到2分钟粒度的时间桶中
• 汇总每个时间桶内各API的错误计数

期望结果 #

格式一（嵌套数组形式） #

[
  {
    _time: ISODate("2022-03-22T00:00:00.000Z"),
    errors: [
      { api: "shipping", count: 30 },
      { api: "inventory", count: 300 }
    ]
  },
  {
    _time: ISODate("2022-03-22T00:02:00.000Z"),
    errors: [
      { api: "inventory", count: 700 },
      { api: "account", count: 1 }
    ]
  }
]

格式二（动态字段形式） #

[
  {
    _time: ISODate("2022-03-22T00:00:00.000Z"),
    shipping: 30,
    inventory: 300
  },
  {
    _time: ISODate("2022-03-22T00:02:00.000Z"),
    inventory: 700,
    account: 1
  }
]

解决方案 #

实现格式一（嵌套数组） #

通过展开错误数组、分组聚合、再重组结构来实现：

db.collection.aggregate([
  // 展开每个文档中的errors数组，每个API错误项成为独立文档
  { $unwind: "$errors" },
  // 按时间桶和API名称分组，汇总计数
  {
    $group: {
      _id: {
        timeBucket: {
          $dateTrunc: {
            date: "$_time",
            unit: "minute",
            binSize: 2
          }
        },
        api: "$errors.api"
      },
      totalCount: { $sum: "$errors.count" }
    }
  },
  // 按时间桶重新分组，将同一桶的API错误汇总到数组中
  {
    $group: {
      _id: "$_id.timeBucket",
      errors: {
        $push: {
          api: "$_id.api",
          count: "$totalCount"
        }
      }
    }
  },
  // 重命名_id为_time，整理输出结构
  {
    $project: {
      _id: 0,
      _time: "$_id",
      errors: 1
    }
  }
])

实现格式二（动态字段） #

在上述基础上，使用$arrayToObject将数组转换为动态字段：

db.collection.aggregate([
  { $unwind: "$errors" },
  {
    $group: {
      _id: {
        timeBucket: {
          $dateTrunc: {
            date: "$_time",
            unit: "minute",
            binSize: 2
          }
        },
        api: "$errors.api"
      },
      totalCount: { $sum: "$errors.count" }
    }
  },
  {
    $group: {
      _id: "$_id.timeBucket",
      apiCounts: {
        $push: {
          k: "$_id.api",
          v: "$totalCount"
        }
      }
    }
  },
  // 将apiCounts数组转换为键值对字段
  { $replaceRoot: {
      newRoot: {
        $mergeObjects: [
          { _time: "$_id" },
          { $arrayToObject: "$apiCounts" }
        ]
      }
    }
  }
])

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