使用 mongodb 聚合框架计算频率

Question

我正在尝试根据 10 秒的间隔计算我的数据库中文档的频率。

这是我的数据库对象的样子：

[
  {
     created_at: "2014-03-31T22:30:48.000Z",
     id: 450762158586880000,
     _id: "5339ec9808eb125965f2eae1"
  },
  {
     created_at: "2014-03-31T22:30:48.000Z",
     id: 450762160407597060,
     _id: "5339ec9808eb125965f2eae2"
  },
  {
     created_at: "2014-03-31T22:30:49.000Z",
     id: 450762163482017800,
     _id: "5339ec9908eb125965f2eae3"
  },
  {
     created_at: "2014-03-31T22:30:49.000Z",
     id: 450762166367707140,
     _id: "5339ec9908eb125965f2eae4"
  },
  {
     created_at: "2014-03-31T22:30:50.000Z",
     id: 450762167412064260,
     _id: "5339ec9a08eb125965f2eae5"
  }
]

我已设法在给定的时间间隔内显示频率，但我希望每 10 秒显示一次。所以最好我的 JSON 看起来像：

[
  {
     time_from: "2014-03-31T22:30:48.000Z",
     time_to: "2014-03-31T22:30:58.000Z",
     count: 6
  },
  {
     time_from: "2014-03-31T22:30:58.000Z",
     time_to: "2014-03-31T22:31:08.000Z",
     count: 3
  },
  {
     time_from: "2014-03-31T22:31:08.000Z",
     time_to: "2014-03-31T22:31:18.000Z",
     count: 10
  },
  {
     time_from: "2014-03-31T22:31:18.000Z",
     time_to: "2014-03-31T22:31:28.000Z",
     count: 1
  },
  {
     time_from: "2014-03-31T22:31:28.000Z",
     time_to: "2014-03-31T22:31:38.000Z",
     count: 3
  }
]

这是我到目前为止所做的：

exports.findAll = function (req, res) {
    db.collection(collection_name, function (err, collection) {
        collection.find().toArray(function (err, items) {
            collection.find().sort({"_id": 1}).limit(1).toArray(function (err, doc) {
                var interval = 100000; // in milliseconds
                var startTime = doc[0].created_at;
                var endTime = new Date(+startTime + interval);

                collection.aggregate([
                    {$match: {"created_at": {$gte: startTime, $lt: endTime}}},
                    {$group: {"_id": 1, "count":{$sum: 1}}}
                ], function(err, result){
                    console.log(result);
                    res.send(result);
                });
            });
        })
    });
};

结果如下：

[
  {
     _id: 1,
     count: 247
  }
]

编辑：

collection.aggregate([
                    { $group: {
                        _id: {
                            year: { '$year': '$created_at'},
                            month: {'$month': '$created_at'},
                            day: {'$dayOfMonth': '$created_at'},
                            hour: {'$hour': '$created_at'},
                            minute: {'$minute': '$created_at'},
                            second: {'$second': '$created_at'}
                        },
                        count: { $sum : 1 }
                    } }
                ], function (err, result) {
                    console.log(result);
                    res.send(result);
                });

导致：

[
  {
     _id: {
        year: 2014,
        month: 3,
        day: 31,
        hour: 22,
        minute: 37,
        second: 10
     },
     count: 6
  }, ...

新的进度，现在如何以 10 秒的间隔显示？

Answer 1

如果只是在 10 秒的时间间隔内获取数据，您可以做一些数学运算并通过聚合运行：

db.collection.aggregate([
    { "$group": {
        "_id": {
             "year": { "$year": "$created_at" },
             "month":{ "$month": "$created_at" },
             "day": { "$dayOfMonth": "$created_at" },
             "hour": { "$hour": "$created_at" },
             "minute": { "$minute": "$created_at" },
             "second": { "$subtract": [
                 { "$second": "$created_at" },
                 { "$mod": [
                     { "$second": "$created_at" },
                     10
                 ]}
             ]}
        },
        "count": { "$sum" : 1 }
    }}
])

因此，这会将事情分解为每分钟 10 秒的时间间隔，并通过一点 mod 10 数学运算发生。

我认为这是合理的，并且由于它使用聚合将是最快的跑步者。如果你真的需要你的序列从最初匹配的时间开始运行 10 秒，那么你可以使用 mapReduce 来完成这个过程：

首先是一个映射器：

var mapper = function () {

    if ( this.created_at.getTime() > ( last_date + 10000 ) ) {
        if ( last_date == 0 ) {
            last_date = this.created_at.getTime();
        } else {
            last_date += 10000;
        }
    }

    emit(
        {
            start: new Date( last_date ),
            end: new Date( last_date + 10000 )
        },
        this.created_at
    );

}

所以这将在 10 秒的间隔内发出日期，从第一个日期开始，然后在每次发现超出范围时增加间隔

现在你需要一个减速器：

var reducer = function (key, values) {
    return values.length;
};

非常简单。只需返回传入的数组长度即可。

因为 mapReduce 以它的方式工作，任何不超过一个值的东西都不会传递给 reducer，所以用 finalize 清理它：

var finalize = function (key, value) {
    if ( typeof(value) == "object" ) {
        value = 1;
    }
    return value;
};

然后只需运行它即可获得结果。请注意传递要在映射器中使用的全局变量的“范围”部分：

db.collection.mapReduce(
    mapper,
    reducer,
    { 
        "out": { "inline": 1 }, 
        "scope": { "last_date": 0 }, 
        "finalize": finalize 
    }
)

每种方法的结果可能略有不同，但这就是重点。这取决于您实际要使用哪一个。

---

考虑到您的评论，您可以“检查”任一语句的输出并以编程方式“填补空白”。我通常更喜欢该选项，但这不是我的程序，我不知道您尝试从该查询中检索多大的系列。

在服务器端，您可以修补“映射器”以执行以下操作：

var mapper = function () {

    if ( this.created_at.getTime() > ( last_date + 10000 ) ) {

        if ( last_date == 0 ) {
            last_date = this.created_at.getTime();
        } else {
            // Patching for empty blocks
            var times = Math.floor( 
                 ( this.created_at.getTime() - last_date ) / 10000
            );

            if ( times > 1 ) {
                for ( var i=1; i < times; i++ ) {
                    last_date += 10000;
                    emit(
                        {
                            start: new Date( last_date ),
                            end: new Date( last_date + 10000 )
                        },
                        0
                    );
                }
            }
            // End patch
            last_date += 10000;
        }
    }

    emit(
        {
            start: new Date( last_date ),
            end: new Date( last_date + 10000 )
        },
        this.created_at
    );

}