MongoDB聚合管道：计算时间序列数据中特定字段的逐小时差值-js教程-PHP中文网

MongoDB聚合管道：计算时间序列数据中特定字段的逐小时差值

本教程详细阐述如何利用MongoDB聚合管道计算时间序列数据中特定字段（如能源消耗）的逐小时差值。通过组合$sort、$group、$setWindowFields等阶段，文章演示了如何针对不同类别（如设备编码）高效地提取每小时的首个记录值，并计算当前小时与前一小时之间该字段的增量，适用于监控系统、物联网数据分析等场景。

在处理时间序列数据时，我们经常需要分析某个指标在不同时间段内的变化量，例如计算每小时的能源消耗增量。本教程将指导您如何使用mongodb的聚合管道（aggregation pipeline）来实现这一目标，特别是如何针对具有不同分类（如设备代码）的数据计算逐小时的字段差值。

假设我们有如下的能源消耗数据，其中包含时间戳（timestamp）、设备代码（code）和能源读数（energy）：

[
  { "_id": 1, "timestamp": "2023-05-15T10:00:00Z", "code": "abc", "energy": 2333 },
  { "_id": 2, "timestamp": "2023-05-15T10:10:00Z", "code": "abc", "energy": 2340 },
  { "_id": 3, "timestamp": "2023-05-15T10:30:00Z", "code": "abc", "energy": 2349 },
  { "_id": 4, "timestamp": "2023-05-15T10:40:00Z", "code": "abc", "energy": 2355 },
  { "_id": 5, "timestamp": "2023-05-15T10:50:00Z", "code": "abc", "energy": 2360 },
  { "_id": 6, "timestamp": "2023-05-15T11:00:00Z", "code": "abc", "energy": 2370 },
  { "_id": 7, "timestamp": "2023-05-15T10:00:00Z", "code": "def", "energy": 3455 },
  { "_id": 8, "timestamp": "2023-05-15T10:10:00Z", "code": "def", "energy": 3460 },
  { "_id": 9, "timestamp": "2023-05-15T10:30:00Z", "code": "def", "energy": 3470 },
  { "_id": 10, "timestamp": "2023-05-15T10:40:00Z", "code": "def", "energy": 3480 },
  { "_id": 11, "timestamp": "2023-05-15T10:50:00Z", "code": "def", "energy": 3490 },
  { "_id": 12, "timestamp": "2023-05-15T11:00:00Z", "code": "def", "energy": 3500 }
]

登录后复制

我们的目标是计算每个code在每个小时开始时的energy读数与前一小时开始时的energy读数之间的差值。例如，对于code: "abc"，我们需要计算11:00的energy（2370）减去10:00的energy（2333），得到差值37。

聚合管道实现步骤

以下是实现上述目标的MongoDB聚合管道详细步骤：

1. 准备数据：初始排序 ($sort)

在进行后续的分组操作之前，首先对数据按时间戳进行升序排序，这确保了$first操作能够准确地获取到每个小时内的第一个energy读数。

{ $sort: { timestamp: 1 } }

登录后复制

2. 按小时和类别分组并提取首个值 ($group)

这一步是核心，我们将数据按code和timestamp截断到小时进行分组。对于每个分组，我们提取该小时内该code的第一个energy读数。$dateTrunc操作符用于将日期截断到指定的单位（此处为"hour"）。

{
  $group: {
    _id: {
      hour: { $dateTrunc: { date: "$timestamp", unit: "hour" } },
      code: "$code"
    },
    firstEnergyInHour: { $first: "$energy" }
  }
}

登录后复制

执行此阶段后，我们将得到类似如下的中间结果：

[
  { "_id": { "hour": ISODate("2023-05-15T10:00:00Z"), "code": "abc" }, "firstEnergyInHour": 2333 },
  { "_id": { "hour": ISODate("2023-05-15T11:00:00Z"), "code": "abc" }, "firstEnergyInHour": 2370 },
  { "_id": { "hour": ISODate("2023-05-15T10:00:00Z"), "code": "def" }, "firstEnergyInHour": 3455 },
  { "_id": { "hour": ISODate("2023-05-15T11:00:00Z"), "code": "def" }, "firstEnergyInHour": 3500 }
]

登录后复制

3. 再次排序以便窗口函数处理 ($sort)

为了确保后续的$setWindowFields操作能够正确地识别前一个小时的数据，我们需要对上一步分组后的结果进行排序。排序的顺序是先按code，再按hour。这样可以保证在每个code组内，小时是按升序排列的。

{
  $sort: {
    "_id.code": 1,
    "_id.hour": 1
  }
}

登录后复制

4. 使用窗口函数获取前一个小时的值 ($setWindowFields)

$setWindowFields是MongoDB 5.0+引入的强大功能，允许我们在一个“窗口”内执行聚合操作。

序列猴子开放平台

具有长序列、多模态、单模型、大数据等特点的超大规模语言模型

查看详情

partitionBy: "$_id.code"：这指定了窗口操作的分区键。这意味着我们将对每个独立的code组执行窗口计算，确保不同设备的数据不会混淆。
sortBy: { "_id.hour": 1 }：在每个分区内，数据将按小时升序排列。
output: { prevEnergy: { $push: "$firstEnergyInHour", window: { documents: [-1, 0] } } }：
- prevEnergy是新创建的字段，它将包含一个数组。
- $push: "$firstEnergyInHour"：将当前文档的firstEnergyInHour值推入数组。
- window: { documents: [-1, 0] }：定义了窗口的范围。[-1, 0]表示当前文档（0）和其前一个文档（-1）。因此，prevEnergy数组将包含当前小时的firstEnergyInHour和前一小时的firstEnergyInHour。

{
  $setWindowFields: {
    partitionBy: "$_id.code",
    sortBy: { "_id.hour": 1 },
    output: {
      prevEnergy: {
        $push: "$firstEnergyInHour",
        window: { documents: [-1, 0] }
      }
    }
  }
}

登录后复制

经过此阶段，数据将变为：

[
  // ... (for code: "abc")
  { "_id": { "hour": ISODate("2023-05-15T10:00:00Z"), "code": "abc" }, "firstEnergyInHour": 2333, "prevEnergy": [2333] }, // 第一个小时，没有前一个值
  { "_id": { "hour": ISODate("2023-05-15T11:00:00Z"), "code": "abc" }, "firstEnergyInHour": 2370, "prevEnergy": [2333, 2370] },
  // ... (for code: "def")
  { "_id": { "hour": ISODate("2023-05-15T10:00:00Z"), "code": "def" }, "firstEnergyInHour": 3455, "prevEnergy": [3455] },
  { "_id": { "hour": ISODate("2023-05-15T11:00:00Z"), "code": "def" }, "firstEnergyInHour": 3500, "prevEnergy": [3455, 3500] }
]

登录后复制

5. 过滤无效结果 ($match)

对于每个code的第一个小时记录，prevEnergy数组中将只有一个元素（当前小时的值），因为它没有前一个小时的数据。为了只保留可以计算差值的记录，我们过滤掉prevEnergy数组中第二个元素（索引1）不存在的文档。

{ $match: { "prevEnergy.1": { $exists: true } } }

登录后复制

6. 计算差值并格式化输出 ($project)

最后一步是计算差值并格式化输出结果。

_id: 0：排除默认的_id字段。
timestamp: "$_id.hour"：将当前小时的时间戳作为输出的timestamp。
code: "$_id.code"：输出设备代码。
energy: { $subtract: [{ $last: "$prevEnergy" }, { $first: "$prevEnergy" }] }：计算prevEnergy数组中最后一个元素（当前小时的值）减去第一个元素（前一小时的值），得到差值。

{
  $project: {
    _id: 0,
    timestamp: "$_id.hour",
    code: "$_id.code",
    energy: { $subtract: [{ $last: "$prevEnergy" }, { $first: "$prevEnergy" }] }
  }
}

登录后复制

完整的聚合管道代码

将以上所有阶段组合起来，得到完整的MongoDB聚合管道：

db.collection.aggregate([
  // 1. 初始排序，确保$first能取到每小时的第一个值
  { $sort: { timestamp: 1 } },
  // 2. 按小时和设备代码分组，并获取每小时的第一个能源读数
  {
    $group: {
      _id: {
        hour: { $dateTrunc: { date: "$timestamp", unit: "hour" } },
        code: "$code"
      },
      firstEnergyInHour: { $first: "$energy" }
    }
  },
  // 3. 再次排序，为$setWindowFields准备数据，确保同一code下小时有序
  { $sort: { "_id.code": 1, "_id.hour": 1 } },
  // 4. 使用窗口函数获取当前小时和前一小时的能源读数
  {
    $setWindowFields: {
      partitionBy: "$_id.code", // 按设备代码分区
      sortBy: { "_id.hour": 1 }, // 在分区内按小时排序
      output: {
        prevEnergy: {
          $push: "$firstEnergyInHour",
          window: { documents: [-1, 0] } // 窗口包含前一个文档和当前文档
        }
      }
    }
  },
  // 5. 过滤掉没有前一个小时数据的记录（即每个code的第一个小时记录）
  { $match: { "prevEnergy.1": { $exists: true } } },
  // 6. 计算差值并格式化输出
  {
    $project: {
      _id: 0,
      timestamp: "$_id.hour", // 输出当前小时的时间戳
      code: "$_id.code",
      energy: { $subtract: [{ $last: "$prevEnergy" }, { $first: "$prevEnergy" }] }
    }
  }
])

登录后复制

执行上述管道后，您将获得预期的输出：

[
  { "timestamp": "2023-05-15T11:00:00Z", "code": "abc", "energy": 37 },
  { "timestamp": "2023-05-15T11:00:00Z", "code": "def", "energy": 45 }
]

登录后复制

注意事项

时间戳数据类型： 确保您的timestamp字段是MongoDB的ISODate类型，而不是字符串。如果它是字符串，您可能需要在聚合管道的早期阶段使用$toDate将其转换为日期类型。
数据完整性： 如果某个小时内没有数据，那么该小时将不会出现在$group阶段的输出中，因此也不会参与后续的差值计算。如果需要处理这种情况（例如，将缺失值视为0），则需要更复杂的聚合逻辑，可能涉及$unionWith或在应用程序层面进行数据填充。
时区： $dateTrunc默认使用UTC时间。如果您的时间戳涉及特定时区，请确保在$dateTrunc中使用timezone选项进行正确的处理。
性能考量： 对于非常大的数据集，聚合管道的性能至关重要。
- 确保timestamp字段上有索引，可以加速$sort和$group操作。
- $setWindowFields在处理大量数据时可能会消耗较多内存。考虑在管道早期使用$match来限制处理的数据量，例如只查询特定日期范围的数据。
- MongoDB 5.0+的聚合管道优化功能会尽可能地将阶段下推到查询层，以提高效率。