Aggregieren und aktualisieren Sie MongoDB

Nutzen Sie für eine verbesserte Leistung, insbesondere beim Umgang mit großen Sammlungen, die Verwendung von Bulk() API für Massenaktualisierungen, da Sie die Vorgänge in Stapeln an den Server senden (z. B. eine Stapelgröße von 1000), was Ihnen eine viel bessere Leistung bietet, da Sie nicht jede Anfrage an den Server senden (wie Sie es derzeit tun). tun mit der Update-Anweisung innerhalb des forEach() Schleife), sondern nur einmal alle 1000 Anfragen, wodurch Ihre Aktualisierungen effizienter und schneller als derzeit sind.

Die folgenden Beispiele demonstrieren diesen Ansatz, das erste verwendet den Bulk() API verfügbar in den MongoDB-Versionen >= 2.6 and < 3.2 . Es aktualisiert alle Dokumente in den clients Abholung, indem Sie nb_orders_1year ändern Felder mit Werten aus den Aggregationsergebnissen.

Seit dem aggregate() -Methode gibt einen cursor zurück , Sie können forEach() Methode, um es zu iterieren und auf jedes Dokument zuzugreifen, wodurch die Massenaktualisierungsvorgänge in Stapeln eingerichtet werden, um sie dann effizient mit der API über den Server zu senden:

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

Das nächste Beispiel gilt für die neue MongoDB-Version 3.2 die seit den Bulk verworfen hat API und einen neueren Satz von APIs mit bulkWrite() .

Es verwendet den gleichen Cursor wie oben, aber anstatt das Ergebnis zu iterieren, erstellen Sie das Array mit den Massenoperationen mithilfe seines map() Methode:

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