计算MapReduce中最后X个文档的平均值

Question

我甚至知道mapreduce是否是我所需要的最佳选项。我有一个猫鼬文件，像这样：

currency { 
   Time: Date,
   Interval: Number
}

在我的mapreduce工作中，我想计算最后X个文档的平均价格（间隔）。（包括电流）。

如果我把20传给我的方法，我想为每个文档计算最后一个19+当前一个的值，除以20。任何正确方向的建议或指示都将不胜感激。这就是我正在努力实现的目标：

function calculateAverages(Schema, interval, avg, callback){
    v
    var o = {};
    o.scope = {interval: interval, avg: avg};
    o.map = function(){
            var value = {
                  Time: this.Date,
                  Interval: this.Interval
            };
            // How am I gonna group the correct number of docs togheter?
            var key= ??
            emit(key, value);

    };
    //  an array of avg ( ex 20) number of items should be passed here
    o.reduce = function(key, intervals){
            var reducedVal = { avg: 0};
            for(var i=0;i<intervals.length;i++){
                reducedVal.avg += intervals[i].Interval;
            }
        reducedVal.avg /= avg;
        return reducedVal;
    };
    o.out = {
      merge: "testing"
    };
    o.finalize = function(key, reducedVal){
        return reducedVal;
    };
    Schema.mapReduce(o, function (err, results) {
        if (err) throw err;
        //console.log(results);
        console.log("mapReduce complete");
        callback(results);
    });
};

Answer 1

您可以使用一个简单的聚合管道来获得平均值，管道将遵循以下步骤

1. 按"时间"字段对文档进行排序
2. 限制文档数量
3. 将文档分组并获得平均值

您可以尝试下面的代码，只需创建Currency模型并向mongodb发送聚合查询，结果就会显示在控制台上，您的集合应该是mongodb服务器上的nammed Currency。

var mongoose = require('mongoose');
var db = mongoose.connection;
mongoose.connect('mongodb://localhost/test');
var CurrencySchema = mongoose.Schema({
    Time: Date,
    Interval: Number
}, {collection: 'currency'});
var Currency = mongoose.model('Currency', CurrencySchema);
// You can change the $limit to specify the number of document
db.once('open', function (callback) {
  var pipeline = [
    {$sort: {Time : -1 }},
    {$limit: 5},
    {$group: {_id: null, average : {$avg: "$Interval"}}}
  ];
  Currency.aggregate(pipeline).exec(function(err, data) {
     console.log(data);
  });
});

Answer 2

如果你想继续使用MapReduce，我的第一种方法是使用"GROUP"+X作为键，其中X是你在每组之后递增的数值。在您可以访问的任何位置声明n=1，并用n++递增，当n%avg=0时，则在您的发射之后X++/n=1。

很抱歉，如果我用Java键入我的示例，那就是我在MapReduce中使用的语言。但我认为你很容易理解我的方法：

private int X = 1, n=1;
public map(Text anyKey, YourStructure value){
    int avg = ....; //Recover your avg value here, wherever it is stored.
    emit("GROUP"+X, value);
    if (n % avg == 0){
        X++;
        n=1;
    }else{
        n++;
    }
}

你发射：

Doc 1 --> key: GROUP1, value: {.....}
Doc 2 --> key: GROUP1, value: {.....}
...
Doc 19 --> key: GROUP1, value: {.....}
Doc 20 --> key: GROUP1, value: {.....}
Doc 21 --> key: GROUP2, value: {.....}
Doc 22 --> key: GROUP2, value: {.....}
...
Doc 29 --> key: GROUP2, value: {.....}
Doc 30 --> key: GROUP2, value: {.....}
Doc 31 --> key: GROUP3, value: {.....}
...

通过这种方式，您将收到您想要在降价中计算平均价格的文件数量。

希望这对你有帮助。