Не удается запустить скрипт оболочки mongo на несколько миллионов данных

Question

У меня есть скрипт в mongoshell, который должен заполнить коллекцию (dataaggregation) из другой (данные), объединив таймеры на каждые 5 минут.
Сбор данных имеет 7.000.000+ записей, и сценарий занимает много времени, чтобы завершить ... 8 часов за 500 000 данных, которые необходимо учитывать и теперь кажется замороженным.

В основном сбор данных содержит записи, как:

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

Коллекция dataaggreagtion содержит записи, как:

{ 
    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 
} 

Каждая запись сбора данных будет являться частью записи о dataaggregation коллекция (будет считаться 1 в атрибуте count).

// 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} 
    ); 
    } 
}) 

Любая идея или предложение улучшить это? Есть ли что-то очевидное, что мне не хватает?

Answer 1

Почему бы не просто использовать aggregation framework для этого? Трубопровод $group делает это вместе с other operators для обработки ваших расчетов.

Для этого вам, вероятно, понадобится версия сервера MongoDB версии 2.6 или выше. Я бы предложил запустить это, включив опцию allowDiskUse, используя этап конвейера $out, чтобы написать коллекцию.

Первое, что вам нужно сделать здесь, - это преобразовать все ваши «строковые» данные в датах в реальные Date объектов. Это довольно просто сделать и хорошо ссылается здесь на StackOverflow, поскольку это общая ошибка моделирования.

Возможно, самый простой способ сделать это - с помощью базовой «математики даты». Объекты Date в MongoDB реагируют на математические операции с другими объектами даты, возвращая значение «timestamp» эпохи (при вычитании из самой даты эпохи, иначе это просто число с разницей в миллисекундах). Это делает интервалы просто:

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 }) 

Long многословны, но это зависит от мощности вы хотите.В любом случае основной подход заключается в использовании оператора modulo $mod, чтобы определить интервалы в 5 минут от числовых результатов, представленных в каждом случае.

В любом случае вы не получите объект Date в результате, но вы получите то, что можно легко «отличить» в объект Date.

Возможно, вы захотите опустить трубопровод $project, если вы можете жить с коллекцией «dataaggregation», имеющей составное поле _id вместо отдельных полей для «t» и «pid», чтобы сделать это еще более эффективным.

Пока ваш сервер может справиться с этим, он будет работать намного быстрее, чем передача клиенту и запись в базу данных.

---

Просто для того, здесь подход к «слепок» все строки, даты, используя 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(); 

Answer 2

Трудно сказать, почему это происходит медленно, но несколько вещей, которые я заметил/сделал бы по-другому:

• Использование $ вкл вместо $ установлен на приращение счетчика и просуммировать

• Создать комбинированный индекс по т и Pid

вы могли бы также рассмотреть вопрос о чтении данных заказанных ISODate, а затем только написать 5 минут ведра с MongoDB как только вы достигнете новых 5 минут ведра , Это значительно уменьшит количество чтений и записей в вашей совокупной коллекции.

Answer 3

Как предложил Нил, я предлагаю вам использовать агрегацию Монго. Если вы хотите, чтобы агрегировать данные все 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 })