基于MongoDB的IoT时序数据秒级汇总与补全方案问询

背景与数据结构 #

我们拥有数千台IoT设备，全天持续发送测量数据，这些数据以简单时序结构存储在MongoDB中，每条记录包含数值、时间戳和设备标识符，示例结构如下：

{
  "observed_at": {
    "$date": "2024-04-03T16:52:56.217Z"
  },
  "meta": {
    "snid": "xyz",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "....."
  },
  "value": 100
}

需求：生成指定日期内每秒总消耗报告 #

需要计算指定日期内每秒所有设备的消耗总和，计算逻辑为：设备在两次上报时间之间，持续使用最近一次上报的数值作为每秒消耗值。示例数据与计算逻辑如下：

DeviceA:
16:52:56 : 100watt
17:01:15 : 0watt

DeviceB:
16:56:30 : 50watt
16:58:38 : 100watt
17:00:04 : 0watt

• DeviceA在16:52:56至17:01:15期间每秒消耗100watt；
• DeviceB在16:56:30至16:58:38期间每秒消耗50watt，之后至17:00:04期间每秒消耗100watt。

最终计算结果示例：

16:52:56 : 100watt
16:52:57 : 100watt
...
16:56:30 : 150watt
16:56:31 : 150watt
...
16:58:38 : 200watt
16:58:39 : 200watt
...
17:00:04 : 100watt
17:00:05 : 100watt
...
17:01:15 : 100watt
17:01:16 : 0watt

初始方案（$densify + $fill）的性能瓶颈 #

最初考虑使用$densify补全每秒时间点，再用$fill填充缺失值，最后按时间分组求和，但很快遇到性能问题：

• 若有5000台设备，每日计算会生成5000*60*60*24=4.32亿文档，数据量随设备数量增长线性上升；
• $densify默认限制仅能生成50万文档，即使拆分处理也不具备可扩展性。

对应的聚合管道示例：

[
  {
    $densify:
      /*
       * 按秒补全指定时间段内的时间点
       */
      {
        field: "from",
        partitionByFields: ["snid"],
        range: {
          step: 1,
          unit: "second",
          bounds: [
            ISODate("2024-05-23T01:00:00Z"),
            ISODate("2024-05-23T01:01:00Z") // 仅示例1分钟区间
          ]
        }
      }
  },
  {
    $fill:
      /*
       * 使用LOCF（最后一次观测值向前填充）补全缺失值
       */
      {
        partitionByFields: ["snid"],
        sortBy: {
          from: 1
        },
        output: {
          value: {
            method: "locf"
          }
        }
      }
  },
  {
    $group:
      {
        _id: "$from",
        sum_power_w: {
          $sum: "$value"
        }
      }
  }
]

备选方案：基于时间区间的转换 #

另一种思路是用$shift将每条设备记录转换为时间区间+对应数值的结构，代表该数值生效的时间范围，示例结构如下：

// 仅保留时分秒的时间戳示例
{
 from: null,
 until: "16:56:30",
 value: 0
}
{
 from: "16:56:30",
 until: "16:58:38",
 value: 50
},
{
 from: "16:58:38",
 until: "17:00:04",
 value: 100
},
{
 from: "17:00:04",
 until: null,
 value: 0
}

目前已实现部分聚合管道用于生成时间区间，后续需完成计算逻辑：

[
  ...,
  {
    $setWindowFields:
      /*
       * 计算每条记录的结束时间戳
       */
      {
        partitionBy: "$meta.snid",
        sortBy: {
          "observed_at": 1
        },
        output: {
          until: {
            $shift: {
              output: "$observed_at",
              by: 1
            }
          }
        }
      }
  }
]

核心问题 #

1. 如何在MongoDB中处理这种大规模的插值计算？
2. 第二种基于时间区间的方案能否完全在MongoDB内实现（无需应用层参与计算）？

补充样本数据 #

[{
  "observed_at": {
    "$date": "2024-05-23T16:53:40.396Z"
  },
  "meta": {
    "snid": "XYZ123456",
    "type": "total_charge_power"
  },
  "value": 1273.979,
  "_id": {
    "$oid": "664f749a092659a0b468914e"
  }
},
{
  "observed_at": {
    "$date": "2024-05-23T16:53:04.253Z"
  },
  "meta": {
    "snid": "XYZ654321",
    "type": "total_charge_power"
  },
  "value": 2204.606,
  "_id": {
    "$oid": "664f7472092659a0b4689134"
  }
},
{
  "observed_at": {
    "$date": "2024-05-23T16:53:12.272Z"
  },
  "meta": {
    "snid": "XYZ654321",
    "type": "total_charge_power"
  },
  "value": 2146.411,
  "_id": {
    "$oid": "664f747a092659a0b468913e"
  }
},
{
  "observed_at": {
    "$date": "2024-05-23T16:54:33.282Z"
  },
  "meta": {
    "snid": "XYZ654321",
    "type": "total_charge_power"
  },
  "value": 2048.367,
  "_id": {
    "$oid": "664f74cb092659a0b4689166"
  }
},
{
  "observed_at": {
    "$date": "2024-05-23T16:57:41.292Z"
  },
  "meta": {
    "snid": "XYZ654321",
    "type": "total_charge_power"
  },
  "value": 1850.317,
  "_id": {
    "$oid": "664f7587092659a0b46891af"
  }
},
{
  "observed_at": {
    "$date": "2024-05-23T16:53:10.731Z"
  },
  "meta": {
    "snid": "XYZ321123",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "664f7477092659a0b468913a"
  },
  "value": 0
},
{
  "observed_at": {
    "$date": "2024-05-23T16:54:02.184Z"
  },
  "meta": {
    "snid": "XYZ321123",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "664f74aa092659a0b4689153"
  },
  "value": 1218.115
},
{
  "observed_at": {
    "$date": "2024-05-23T16:54:07.244Z"
  },
  "meta": {
    "snid": "XYZ321123",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "664f74af092659a0b4689157"
  },
  "value": 4309.275
},
{
  "observed_at": {
    "$date": "2024-05-23T16:54:12.304Z"
  },
  "meta": {
    "snid": "XYZ321123",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "664f74b4092659a0b4689159"
  },
  "value": 8790.988
},
{
  "observed_at": {
    "$date": "2024-05-23T16:54:17.364Z"
  },
  "meta": {
    "snid": "XYZ321123",
    "type": "total_charge_power"
  },
  "_id": {
    "$oid": "664f74b9092659a0b468915c"
  },
  "value": 10662.279
}]

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