无法在数百万条数据上运行 mongo shell 脚本答案

【问题标题】：Cannot run mongo shell script on several million of data无法在数百万条数据上运行 mongo shell 脚本
【发布时间】：2015-01-08 20:33:53
【问题描述】：

我在 mongoshell 中有一个脚本，它应该从另一个（数据）中填充一个集合（数据聚合），每 5 分钟聚合一次时间序列。
数据收集有 7.000.000 多个条目，脚本需要很长时间才能完成...... 500.000 个数据需要 8 小时才能考虑在内，现在似乎已冻结。

基本上数据集合有这样的记录：

{
  isodate: '2014-12-1OT12:47:32.000+02.00',
  value: 234,
  parentID: 123      
}

数据聚合集合具有如下记录：

{
   t: '2014-12-1OT12:45:00.000+02.00',
   pid: 123,  // parentID
   sum: 1234, // sum of all the value of data between 12:45 and 12:50
   count: 5,  // number of data elements between 12:45 and 12:50
   min: 23,   
   max: 435
}

数据集合的每条记录都将是数据聚合集合记录的一部分（在 count 属性中计为 1）。

// Cleanup collection
db.dataaggregation.remove({})

// Loop through data and populate the dataaggregation collection
db.data.find().addOption(DBQuery.Option.noTimeout).forEach(function(dt){
  // Get 5 minutes timestamp
  // eg: '2014-12-1OT12:47:32.000+02.00' => '2014-12-1OT12:45:00.000+02.00'
  dt.isodate.setMinutes(dt.isodate.getMinutes() - dt.isodate.getMinutes() % 5);
  dt.isodate.setSeconds(0);

  // Create the dataaggregation record for the (timestamp, parentID) couple if does
  // not exist or update the existing one
  var d = db.dataaggregation.findOne({t: dt.isodate, pid: dt.parentID});
  if(!d){
    db.dataaggregation.insert({
        t:dt.isodate,
        pid: dt.parentID,
        sum: dt.value,
        count: 1,
        min: dt.value,
        max: dt.value
    });
  }else{
    db.dataaggregation.update({
        t:dt.isodate,
        pid: dt.parentID
    },{
        $set:{
            sum: d.sum + dt.value,
            count: d.count + 1,
            min: dt.value < d.min ? dt.value : d.min,
            max: dt.value > d.max ? dt.value : d.max
        }
    },
    {upsert:true}
    );
  }
})

有什么想法或建议来改进这一点吗？我有什么明显的遗漏吗？

【问题讨论】：

标签： mongodb mongodb-query aggregation-framework mongo-shell

【解决方案1】：

为什么不直接使用aggregation framework 呢？ $group 管道与 other operators 一起执行此操作以处理您的计算。

您可能需要 MongoDB 2.6 或更高版本的服务器才能执行此操作。我建议运行此程序，打开“allowDiskUse”选项并使用$out 管道阶段写入集合。

您需要在这里做的第一件事是将日期中的所有“字符串”数据转换为真正的Date 对象。这很容易做到，并且在 StackOverflow 上很好地引用了它，因为它是一个常见的建模错误。

可能最简单的方法是使用基本的“日期数学”。 MongoDB 中的日期对象通过返回一个“纪元时间戳”值来响应针对其他日期对象的数学运算（当从纪元日期本身中减去时，否则它只是一个具有毫秒差异的数字）。这使间隔变得简单：

db.data.aggregate([
    { "$group": {
        "_id": {
            "t": {
                "$subtract": [
                    { "$isoDate", new Date("1970-01-01") },
                    { "$subtract": [
                       { "$isoDate", new Date("1970-01-01") },
                       { "$mod": [
                           { "$isoDate", new Date("1970-01-01") },
                           1000 * 60 * 5
                       ]}
                    ]}                          
                ]
            },
            "pid": "$parentID"
        },
        "sum": { "$sum": "$value" },
        "count": { "$sum": 1 },
        "min": { "$min": "$value" },
        "max": { "$max": "$value" }
    }},
    { "$project": {
        "_id": 0,
        "t": "$_id.t",
        "pid": "$_id.pid",
        "sum": 1,
        "count": 1,
        "min": 1,
        "max": 1
    }},
    { "$out": "dataaggregation" }
],{ "allowDiskUse": true })

或者使用日期聚合运算符使用类似的操作：

db.data.aggregate([
    { "$group": {
        "_id": {
            "t": {
                "year": { "$year": "$isodate" },
                "month": { "$month": "$isodate" },
                "dayOfMonth": { "$dayOfMonth": "$isodate" },
                "hour": { "$hour": "$isodate" },
                "minute": {
                    "$mod": [
                        { "$minute": "$isodate" },
                        5
                    ]
                }
            },
            "pid": "$parentID"
        },
        "sum": { "$sum": "$value" },
        "count": { "$sum": 1 },
        "min": { "$min": "$value" },
        "max": { "$max": "$value" }
    }},
    { "$project": {
        "_id": 0,
        "t": "$_id.t",
        "pid": "$_id.pid",
        "sum": 1,
        "count": 1,
        "min": 1,
        "max": 1
    }},
    { "$out": "dataaggregation" }
],{ "allowDiskUse": true })

冗长，但这取决于您想要的输出。在这两种情况下，基本方法是使用取模 $mod 运算符，以便根据每种情况下提供的数字结果确定 5 分钟间隔。

无论哪种方式，您都不会在结果中返回 Date 对象，但您确实会得到可以轻松“转换”到 Date 对象中的东西。

如果您可以使用具有复合 _id 字段的“dataaggregation”集合来代替“t”和“pid”的单独字段，那么您可以选择删除 $project 管道，以使其更有效.

只要您的服务器可以处理它，那么它的运行速度就会比传输到客户端并写回数据库快得多。

只是为了补充，这是一种使用 Bulk Operations API 将所有字符串“转换”为日期的方法：

var bulk = db.collection.initializeOrderdBulkOp();
var counter = 0;

db.collection.find().forEach(function(doc) {
    bulk.find({ "_id": doc._id })
        .updateOne({ "$set": { "isodate": new Date(doc.isodate) } });
    counter++;

    if( counter % 1000 == 0 ) {
        bulk.execute();
        bulk = db.collection.initializeOrderdBulkOp();
    }  
});

if ( counter % 1000 != 0 )
    bulk.execute();

【讨论】：

谢谢，您将如何对时间戳进行分组，以便将每个数据放入正确的 5 分钟存储桶中？例如 2014-12-1OT12:47:32.000+02.00 应该进入以下引用的数据聚合中： 1. 与原始数据记录相同的 parentID； 2.以下时间戳2014-12-1OT12:45:00.000+02.00
@Luc 这可以通过聚合轻松完成。但平心而论，这不是你在这里问的问题。如果您还有其他问题，那么最好再问一个问题。我会根据每个问题明确说明您的意图，并且 StackExchange 模型通常是每个响应“仅一个问题”。日期数学或日期聚合运算符是提示。如果您无法弄清楚，请“问另一个问题”。只要适合使用聚合而不是客户端编码的响应，这个问题就已经得到解答。
那是我的问题的一个隐藏部分（那是在代码的评论中）但你说得对，但我不是很清楚 :)
@Luc 继续尝试。给出了提示。但是，如果您无法解决或不耐烦，那么我等待您的新问题 :)
嗯，我的验证速度太快了 :) 我运行了脚本，但是我拥有的每个数据条目都有一个数据聚合条目（count=1，sum=min=max='数据的值")。这是正常的，因为 (isodate, parentID) 对在数据收集中是唯一的。我期望的分组实际上是每 5 分钟一次，因此总和、最小值、最大值、计数都会更新为所有具有时间戳的记录那个时间范围。我会更新可能不太详细的标题。

【解决方案2】：

正如 Neil 建议的那样，我建议您使用 Mongo 聚合。如果你想聚合所有 5 分钟的数据，你可以使用：

db.data.aggregate([
    { "$group": {
        "_id": {
            "t": { $subtract: [{ $subtract: [ "$isodate", { $multiply: [{ $mod: [ {$minute:"$isodate"}, 5 ] }, 60*1000]} ] }, { $multiply: [{ $mod: [ {$second:"$isodate"}, 60 ] }, 1000]}]},
            "pid": "$variableID"
        },
        "sum": { "$sum": "$value" },
        "count": { "$sum": 1 },
        "min": { "$min": "$value" },
        "max": { "$max": "$value" }
    }},
    { "$project": {
        "_id": 0,
        "t": "$_id.t",
        "pid": "$_id.pid",
        "sum": 1,
        "count": 1,
        "min": 1,
        "max": 1
    }},
{ "$out": "dataaggregation" }
],{ "allowDiskUse": true })

【讨论】：

【解决方案3】：

很难说为什么它很慢，但有几件事我注意到/会有所不同：

使用 $inc 而不是 $set 来增加计数和总和
在 t 和 pid 上创建组合索引

您还可以考虑读取按 isodate 排序的数据，然后在到达新的 5 分钟存储桶后仅将 5 分钟存储桶写入 MongoDB。这将大大减少对聚合集合的读取和写入量。

【讨论】：