Совокупные и обновление MongoDB

Question

У меня есть 2 коллекции:

• клиентов (6 000 000 документов)
• заказов (50 000 000 документов)

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

Я попытался это:

db.orders.aggregate(
    {$match: 
     { date_order: { $gt: v_date1year } } 
    }, 
    {$group : { 
     _id : "$id_client", 
     count : {$sum : 1} 
    }} , 
    { 
     "$out": "tmp_indicators" 
    } 
) 

db.tmp_indicators.find({}).forEach(function (my_client) { 
    db.clients.update (
     {"id_client": my_client._id}, 
     {"$set": 
      { "nb_orders_1year" : my_client.count } 
     } 
    ) 
}) 

Я должен сделать это 3 раза, 1 для агрегирования прошлого года, 1 за прошедший месяц и 1 за прошедшую неделю. Обработка очень медленная, есть ли у вас представление о том, как выполнить ее лучше?

Answer 1

Для повышения производительности, особенно при работе с большими коллекциями, воспользоваться помощью Bulk() API для оптовых обновлений, как вы будете отправлять операции на сервер в пакетном режиме (например, скажем, размер пакета 1000), который дает вам гораздо лучшую производительность, так как вы не будете отправлять каждый запрос на сервер (как вы сейчас делаете с оператором обновления в цикле), но только раз в каждые 1000 запросов, что делает ваши обновления более эффективными и быстрыми чем в настоящее время.

Следующие примеры демонстрируют этот подход, первый использует Bulk() API доступны в MongoDB версии >= 2.6 and < 3.2. Он обновляет все документы в коллекции clients, изменяя поля nb_orders_1year со значениями из результатов агрегирования.

Поскольку aggregate() метод возвращает cursor, Вы можете использовать forEach() метода выходного коллекции агрегации, чтобы перебирать его и получить доступ к каждому документу, таким образом, Наладке массовых операций обновления в пакетах, чтобы затем отправить через сервер эффективно с API:

var bulk = db.clients.initializeUnorderedBulkOp() 
    pipeline = [ 
     { 
      "$match": { "date_order": { "$gt": v_date1year } } 
     }, 
     { 
      "$group": { 
       "_id": "$id_client", 
       "count": { "$sum" : 1 } 
      } 
     }, 
     { "$out": "tmp_indicators" }   
    ], 
    counter = 0; 
db.orders.aggregate(pipeline); 
db.tmp_indicators.find().forEach(function (doc) {  
    bulk.find({ "_id": doc._id }).updateOne({ 
     "$set": { "nb_orders_1year": doc.count } 
    }); 

    counter++; 
    if (counter % 1000 == 0) { 
     bulk.execute(); // Execute per 1000 operations and re-initialize every 1000 update statements 
     bulk = db.clients.initializeUnorderedBulkOp(); 
    } 
}); 
// Clean up remaining operations in queue 
if (counter % 1000 != 0) { bulk.execute(); } 

следующий пример относится к новой версии MongoDB 3.2 который с тех пор deprecated the Bulk API и при условии, более новый набор API-интерфейсов с использованием bulkWrite().

Он использует тот же курсор, как указано выше, но вместо того, чтобы итерация результат, создать массив с объемными операций с использованием его map() метод:

var pipeline = [ 
     { 
      "$match": { "date_order": { "$gt": v_date1year } } 
     }, 
     { 
      "$group": { 
       "_id": "$id_client", 
       "count": { "$sum" : 1 } 
      } 
     }, 
     { "$out": "tmp_indicators" }   
    ]; 
db.orders.aggregate(pipeline); 
var bulkOps = db.tmp_indicators.find().map(function (doc) { 
     return { 
      "updateOne": { 
       "filter": { "_id": doc._id } ,    
       "update": { "$set": { "nb_orders_1year": doc.count } } 
      }   
     }; 
    }); 

db.clients.bulkWrite(bulkOps, { "ordered": true });