MongoDB如何对嵌套数组按时间间隔求均值并保留原结构

场景说明 #

• 业务采用size bucket模式存储时序数据，需对嵌套samples数组按1/5/10分钟等指定时间间隔计算平均值，实现采样点压缩。
• 原始存储文档字段说明：
  • _id：文档唯一标识
  • data：ObjectId类型，业务数据维度标识
  • day：日期维度字段
  • first：当前桶内最早采样时间
  • last：当前桶内最晚采样时间
  • nSamples：桶内采样点总条数
  • samples：采样点数组，单文档最多存200条记录，每条记录含time（ISODate类型）、value两个字段
  • version：文档版本号
• 原有管道逻辑问题：
  1. $match筛选指定data范围、first时间符合条件的文档
  2. $unwind展开samples数组
  3. 时间戳取模分组后用$avg计算平均值
     存在缺陷：分组键未携带data维度标识，导致跨data混算，且分组后未还原原文档结构，无法直接得到压缩后的桶文档。

调整后聚合管道 #

核心调整点：

• 第一次分组时将*data、原文档唯一标识、公共维度字段*加入分组键，从根源避免跨data、跨原桶混算
• 第二次按原文档维度分组，将聚合后的采样点重新组装为samples数组
• 最后还原原文档结构，同步更新压缩后的first/last/nSamples统计字段

以5分钟（300000毫秒）采样间隔为例，管道代码如下：

// 可自定义采样间隔，单位为毫秒
const sampleInterval = 5 * 60 * 1000;

db.ts_bucket.aggregate([
  // 原有匹配阶段无需调整，替换为实际筛选条件即可
  {
    $match: {
      data: { $in: [ObjectId("业务数据ID1"), ObjectId("业务数据ID2")] },
      first: { $gte: ISODate("2024-01-01T00:00:00Z") }
    }
  },
  // 展开samples数组，可选保留数组索引方便校验
  {
    $unwind: {
      path: "$samples",
      includeArrayIndex: "sampleIndex"
    }
  },
  // 第一层分组：按data+原文档ID+时间桶维度聚合，计算每个时间桶的平均值
  {
    $group: {
      _id: {
        data: "$data",
        originDocId: "$_id",
        day: "$day",
        version: "$version",
        timeSlot: {
          $subtract: [
            { $toLong: "$samples.time" },
            { $mod: [{ $toLong: "$samples.time" }, sampleInterval] }
          ]
        }
      },
      slotAvgValue: { $avg: "$samples.value" }
    }
  },
  // 第二层分组：按原文档维度重组，生成压缩后的samples数组
  {
    $group: {
      _id: {
        _id: "$_id.originDocId",
        data: "$_id.data",
        day: "$_id.day",
        version: "$_id.version"
      },
      compressedSamples: {
        $push: {
          time: { $toDate: "$_id.timeSlot" },
          value: "$slotAvgValue"
        }
      },
      newFirst: { $min: { $toDate: "$_id.timeSlot" } },
      newLast: { $max: { $toDate: "$_id.timeSlot" } }
    }
  },
  // 还原原始文档结构，更新统计字段
  {
    $replaceRoot: {
      newRoot: {
        _id: "$_id._id",
        data: "$_id.data",
        day: "$_id.day",
        version: "$_id.version",
        first: "$newFirst",
        last: "$newLast",
        nSamples: { $size: "$compressedSamples" },
        samples: "$compressedSamples"
      }
    }
  }
])

可选配置说明 #

• 如果不需要保留原始size bucket的文档边界，要跨原桶按data+天维度聚合生成新桶，只需要删除第一层分组里的originDocId字段，第二层分组键仅保留data和day即可
• 如果原文档还有其他需要保留的公共字段，只需要在第一层分组的_id中加入对应字段，第二层分组透传后在$replaceRoot阶段输出即可
• 时间计算前必须通过$toLong将ISODate类型转为毫秒时间戳，否则取模运算会报类型错误
• 压缩比和采样间隔正相关：原200条1秒间隔的采样点，按1分钟间隔压缩后仅保留3~4条记录，按10分钟间隔压缩后仅保留1条记录。

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