MongoDB聚合：如何按open-close状态区间分组文档并计算指标？

针对你的需求，这里提供两种可行的聚合查询方案，分别基于数组遍历处理和窗口函数修正，都能实现将连续的open到后续第一个close的文档分组，并计算平均温度、开启/关闭时间。

方案一：基于数组遍历与$reduce的实现 #

这种方法先将同房间的文档按时间排序后聚合为数组，再通过$reduce遍历数组构建每个完整的open-close周期，适合所有支持MongoDB 3.4+的版本：

db.collection.aggregate([
  // 1. 按房间和时间戳升序排序，确保文档顺序符合时间线
  { $sort: { room: 1, timestamp: 1 } },
  
  // 2. 按房间分组，将该房间所有文档按顺序存入数组
  {
    $group: {
      _id: "$room",
      docs: { $push: "$$ROOT" }
    }
  },
  
  // 3. 遍历数组，构建每个open-close周期
  {
    $project: {
      _id: 0,
      room: "$_id",
      cycles: {
        $reduce: {
          input: "$docs",
          initialValue: { currentCycle: [], result: [] },
          in: {
            $cond: [
              // 遇到open状态文档，加入当前待完成的周期
              { $eq: ["$$this.state", "open"] },
              {
                currentCycle: { $concatArrays: ["$$value.currentCycle", ["$$this"]] },
                result: "$$value.result"
              },
              // 遇到close状态文档，将当前周期+该close文档作为完整周期存入结果，重置当前周期
              {
                currentCycle: [],
                result: {
                  $concatArrays: [
                    "$$value.result",
                    [{ docs: { $concatArrays: ["$$value.currentCycle", ["$$this"]] } }]
                  ]
                }
              }
            ]
          }
        }
      }
    }
  },
  
  // 4. 展开周期数组，每个周期成为独立文档
  { $unwind: "$cycles.result" },
  
  // 5. 计算统计指标并整理输出字段
  {
    $project: {
      room: 1,
      avgTemp: { $avg: "$cycles.result.docs.temp" },
      openTimestamp: { $first: "$cycles.result.docs.timestamp" },
      closeTimestamp: { $last: "$cycles.result.docs.timestamp" }
    }
  }
])

方案二：修正窗口函数的实现 #

针对你最初尝试的$setWindowFields思路，通过调整窗口过滤逻辑，实现为每个文档标记所属周期的close文档ID，再分组统计，适合MongoDB 5.0+版本（支持窗口filter选项）：

db.collection.aggregate([
  // 1. 按房间和时间戳升序排序
  { $sort: { room: 1, timestamp: 1 } },
  
  // 2. 为每个文档标记后续第一个close文档的ID
  {
    $setWindowFields: {
      partitionBy: "$room",
      sortBy: { timestamp: 1 },
      output: {
        closeId: {
          $first: "$_id",
          window: {
            documents: ["current", "unbounded"],
            filter: { $eq: ["$state", "close"] }
          }
        }
      }
    }
  },
  
  // 3. 过滤掉无对应close的open文档（如果存在）
  { $match: { closeId: { $exists: true } } },
  
  // 4. 按房间和closeID分组，计算统计指标
  {
    $group: {
      _id: { room: "$room", closeId: "$closeId" },
      avgTemp: { $avg: "$temp" },
      openTimestamp: { $first: "$timestamp" },
      closeTimestamp: { $max: { $cond: [{ $eq: ["$state", "close"] }, "$timestamp", null] } }
    }
  },
  
  // 5. 整理输出字段
  {
    $project: {
      _id: 0,
      room: "$_id.room",
      avgTemp: 1,
      openTimestamp: 1,
      closeTimestamp: 1
    }
  }
])

说明 #

• 两种方案都能输出你期望的结果，方案一兼容性更好，方案二更贴近你最初的思路。
• 若存在未匹配close的open文档，方案二中的$match阶段会自动过滤这些文档，避免生成无效周期。

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