Mongodb查询获取时间序列的峰值

Question

我有一个 mongodb，里面有一些传感器数据。每次触发传感器时，该值都会增加 1。有时会重置传感器。现在我想将传感器数据的值相加得到一个时间序列的总值。

我的数据如下所示：

{
    value: 10
    timestamp: 1577836800
}, {
    value: 12
    timestamp: 1577836810
}, {
    value: 17 <== PEAK
    timestamp: 1577836820
}, {
    value: 2
    timestamp: 1577836830
}, {
    value: 7
    timestamp: 1577836840
}, {
    value: 10 <== PEAK
    timestamp: 1577836850
},{
    value: 1
    timestamp: 1577836860
}

我的目标是在最小和最大时间戳内获取所有峰值 (17 + 10) 的值，并将它们相加（之后也可以手动完成）。这意味着我必须按时间戳对它们进行排序，并将一个值与前一个值进行比较，以检查哪个值更高。我是 mongo 的新手，不知道从哪里开始。欢迎任何帮助！

Answer 1

我在评论中提到了边缘设备的使用，这使得可以将值分配给适当的时期。使用 nodeJS 服务来确定它必须更新哪个文档，因为它使查询集合变得更加容易和快捷。

您可以在每个文档中使用这种格式：

{
    "period" : {
        "start" : 1577836800,
        "end" : 1577836820
    },
    "values": [
        {
            "value": 10
            "timestamp": 1577836800
        }, 
        {
            "value": 12
            "timestamp": 1577836810
        }, 
        {
            "value": 17
            "timestamp": 1577836820
        }
    ]
}

以传感器为服务提供新值为例。在这种情况下，值为 2，获取最后一个文档并将其与 values 中的最后一个值进行比较。

如果新值小于前一个值，则使用以下信息创建一个新文档，当然不要忘记使用最后一个值的时间戳填写“periode.end”来更新前一个文档。

{
    "period" : {
        "start" : 1577836830,
        "end" : null
    },
    "values": [
        {
            "value": 2
            "timestamp": 1577836830
        }
    ]
}

---

更新

我编写了这个 JavaScript 脚本来从一个集合迁移，根据需求将数据处理成一种新格式，然后将它们插入到新集合中。

function createSensorObject(start, value){
    var doc = new Object();

    doc.period = new Object();
    doc.period["start"] = start;
    doc.period["end"] = null;

    doc.values = new Array();

    var valueObject = new Object();
    valueObject["value"] = value;
    valueObject["timestamp"] = start;

    doc.values.push(valueObject);
    return doc;
}

function pushNewValueIntoSensorObject(doc, value, timestamp){
    doc["period"]["end"] = timestamp;

    var valueObject = new Object();
    valueObject["value"] = value;
    valueObject["timestamp"] = timestamp;

    doc["values"].push(valueObject);
}

function closePeriodInSensorObject(doc) {
    doc["period"]["end"] = doc["period"]["start"];
}

// New collection cursor to insert new values.
var sensorNewCursor = db.sensor_new; 

// Using the aggregate pipeline to sort and using toArray to iterate through.
var sensorDocuments = db.sensor.aggregate([ { $sort: { "timestamp" : 1}}]).toArray();

// Temp array to store the new sensor objects.
var sensorObjectArray = new Array();

for(var i = 0; i < sensorDocuments.length; i++) {
    // Working with the current document.
    var currentDocument = sensorDocuments[i];
    var value = currentDocument["value"];
    var timestamp = currentDocument["timestamp"];

    // Initially creating a new document.
    if(i === 0) {
        var sensorObject = createSensorObject(timestamp, value);

        // If there is only one document then we need to close the document.
        if(i === (sensorDocuments.Length - 1)) {
            closePeriodInSensorObject(sensorObject);
        }

        // Push into temp array.
        sensorObjectArray.push(sensorObject);
    }
    else{   
        // Get the previous value.
        var previousValue = sensorDocuments[i - 1]["value"];

        // Get the latest sensor object from the temp array.
        var sensorObject = sensorObjectArray[sensorObjectArray.length - 1];

        // When the new value is larger or equal to the previous value.
        if(value >= previousValue) {
            // Push the new value into the latest sensor object.
            pushNewValueIntoSensorObject(sensorObject, value, timestamp);
        }
        else {  
            // Create new sensor object.
            var sensorObject = createSensorObject(timestamp, value);
            sensorObjectArray.push(sensorObject);
        }
    }
}

print(`Inserting ${sensorObjectArray.length} objects into the db.sensor_new collection.`);

// Insert the whole array into the new collection
sensorNewCursor.insert(sensorObjectArray);

您可以使用 load 函数从 mongoshell 调用此 js 文件。 https://docs.mongodb.com/manual/reference/method/load/

---

看看两张截图。

---

终于！查询

您必须使用聚合管道来获取每个周期的最大值之和。

首先我们需要展开这些值，然后使用 $max 进行 $group 以找到我们的最大值，然后再次分组但没有 $_id 字段来对这些值进行 $sum。

db.collection.aggregate([
    {
        $unwind: "$values"
    },
    {
        $group: {
            "_id": "$_id",
            "value": {
                $max: "$values.value"
            }
        }
    },
    {
        $group: {
            "_id": null,
            "total": {
                $sum: "$value"
            }
        }
    }
])

这个查询的结果是 28

[
    {
        "_id": null,
        "total": 28
    }
]

MongoPlayground 根据新的文档结构： https://mongoplayground.net/p/WblN5QzPoeB

我真的希望这对您有用，如果没有请发表评论。 :)