MongoDB：有没有办法使用聚合来检测价值趋势？答案

【问题标题】：MongoDB : Is there a way to detect a value trend using aggregation?MongoDB：有没有办法使用聚合来检测价值趋势？
【发布时间】：2015-04-04 19:37:44
【问题描述】：

我正在尝试检测集合中某个值的“趋势”。

假设我有以下内容：

{ created_at: 2014-12-01, value:1015 }
{ created_at: 2014-12-01, value:1015 }
{ created_at: 2014-12-01, value:1019 }
{ created_at: 2014-12-02, value:1018 }
{ created_at: 2014-12-02, value:1021 }
{ created_at: 2014-12-03, value:1010 }
{ created_at: 2014-12-03, value:1012 }
{ created_at: 2014-12-03, value:1011 }
{ created_at: 2014-12-04, value:1012 }

我只想有这样的输出：

{ created_at: 2014-12-01, average: 1016, diff: 0}
{ created_at: 2014-12-02, average: 1019, diff: 3}

其中 diff 是两者的平均值之差两个日期。

我想出了如何计算平均值，找到集合中的最小值/最大值和第一个/最后一个值，但找不到比较两个平均值的方法...

【问题讨论】：

如果在应用程序代码中计算平均差异会更好。目前可能无法使用聚合框架。
您无法像使用聚合框架（一个或多个管道）那样实际计算输出，但是您可以计算很多东西。我认为更多解释您如何以及为什么检测趋势将是帮助我们提供好的建议的最佳方式。
@wdberkeley 我有一个集合，其中包含来自传感器网络（名为“sensor_readings”）的读数。读数以随机时间出现，并以您在示例中看到的方式存储 - 只需添加一个完整的时间戳。我想每 10 分钟聚合一次数据，每分钟对它们进行分组，计算分钟平均值以及与前一分钟的差异。

标签： mongodb mapreduce mongodb-query aggregation-framework

【解决方案1】：

假设这些是根据您的 cmets 在“每分钟聚合”上的实际日期，唯一真正的一次性方法是使用 mapReduce。这里的关键是 mapReduce 可以存储一个全局变量，因此可以“跟踪”您的最后一个结果，以确定每个聚合记录之间的“差异”

db.collection.mapReduce(
    function() {
        // Round date to the minute
        var key = this.created_at.valueOf()
            - ( this.created_at.valueOf() % ( 1000 * 60 ) );
        emit( key, { "average": this.value } );
    },
    function(key,values) {
        values = values.map(function(i) { return i.average });
        var result = {
           "average": Math.floor(Array.avg(values))
        };
        return result;
    },
    {
        "out": { "inline": 1 },
        "scope": { "lastAvg": 0 },
        "finalize": function(key,value) {
            value.diff = ( lastAvg == 0 ) ? 0 : value.average - lastAvg;
            lastAvg = value.average;
            return value;
        }
    }
)

或者，您可以像前面提到的那样“进行后处理”，并在客户端代码中执行相同的操作来计算差异，因为您使用类似的作用域变量迭代游标。以外壳为例：

var lastAvg = 0;
db.collection.aggregate([
    { "$group": {
        "_id": { "$subtract": [
            { "$subtract": [ "$created_date", new Date(0) ] },
            { "$mod": [
                { "$subtract": [ "$created_date", new Date(0) ] },
                1000 * 60
            ]}
        ]},
        "average": { "$avg": "$value" }
    }},
    { "$sort": { "_id": 1 } }
]).forEach(function(doc) {
    doc.average = Math.floor(doc.average);
    doc.diff = ( lastAvg == 0 ) ? 0 : doc.average - lastAvg;
    lastAvg = doc.average;
    printjson(doc);
})

在这两种情况下，我都使用日期数学原理将日期对象转换为 unix/epoch 时间戳表示形式，该数字按分钟四舍五入到最接近的楼层。使用聚合框架，您可以交替使用date aggregation operators 来提取日期部分以进行分组。

在任何一种情况下，将其重新转换为Date 对象都非常简单，需要在内部使用.mapReduce() 或在后期处理中使用.aggregate()。

因此，总而言之，您可以使用 mapReduce 的“全局范围”功能，也可以只处理来自聚合的结果游标，以便计算结果中每个分组之间的差异。

【讨论】：

尼尔，你能告诉我怎么做这个 spring-data-mongodb 吗？

【解决方案2】：

粗略：我会计算十分钟内的平均值：

> var avgCursor = db.sensor_readings.aggregate([
    { "$match" : { "created_at" : { "$gt" : ten_minutes_ago, "$lte" : now } } }
    { "$group" : { "_id" : 0, "average" : { "$avg" : "$value" } } }
]}
> var avgDoc = avgCursor.toArray()[0]
> avgDoc
{ "_id" : 0, "average" : 23 }

然后我会将它存储在另一个集合中：

> db.sensor_averages.insert({ "start" : ten_minutes_ago, "end" : now, "average" : avgDoc.average })

最后，回忆一下计算差值所需的两个平均值，并计算它：

> var diffCursor = db.sensor_averages.find({ "start" : { "$gte" : twenty_minutes_ago } }).sort({ "start" : -1 })
> var diffArray = diffCursor.toArray()
> var difference = diffArray[0].average - diffArray[1].average

您也可以跳过定期聚合，而是在 sensor_averages 中更新运行平均值，每 10 分钟跳转到一个新文档。在每 10 分钟开始时，插入到sensor_averages 一个文档中

{
    "start" : now,
    "svalues" : 0,
    "nvalues" : 0
}

然后在接下来的十分钟内每次插入 sensor_reading 文档时，还要更新 sensor_averages 文档：

db.sensor_averages.update(
    { "start" : now_rounded_to_the_ten_minute_boundary },
    { "$inc" : { "svalues" : value, "nvalues" : 1 } }
)

然后，当您想要平均值之间的差异时，调用相应的两个文档，将svalues 除以nvalues 得到平均值，然后减去。

【讨论】：