本教程详细阐述如何利用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+的聚合管道优化功能会尽可能地将阶段下推到查询层，以提高效率。
通过本教程，您应该能够熟练地使用MongoDB聚合管道计算时间序列数据中特定字段的逐小时差值，并将其应用于您的数据分析和监控需求中。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
