MongoDB
Bulkbewerkingen

Opmerkingen

Een veld naar een ander type converteren en de hele verzameling in bulk bijwerken

Gewoonlijk is het geval wanneer men een veldtype in een ander wil veranderen, bijvoorbeeld kan de originele verzameling "numerieke" of "datum" -velden als strings opslaan:

{
    "name": "Alice",
    "salary": "57871",
    "dob": "1986-08-21"
},
{
    "name": "Bob",
    "salary": "48974",
    "dob": "1990-11-04"
}

Het doel zou zijn om een gigantische verzameling zoals hierboven bij te werken

{
    "name": "Alice",
    "salary": 57871,
    "dob": ISODate("1986-08-21T00:00:00.000Z")
},
{
    "name": "Bob",
    "salary": 48974,
    "dob": ISODate("1990-11-04T00:00:00.000Z")
}

Voor relatief kleine gegevens kan men het bovenstaande bereiken door de verzameling te itereren met behulp van een snapshot met de methode forEach() van de cursor en elk document als volgt bij te werken:

db.test.find({
    "salary": { "$exists": true, "$type": 2 },
    "dob": { "$exists": true, "$type": 2 }
}).snapshot().forEach(function(doc){ 
    var newSalary = parseInt(doc.salary),
        newDob = new ISODate(doc.dob);        
    db.test.updateOne(
        { "_id": doc._id },
        { "$set": { "salary": newSalary, "dob": newDob } }
    );
});

Hoewel dit optimaal is voor kleine collecties, worden de prestaties met grote collecties aanzienlijk verminderd, aangezien het doorlopen van een grote gegevensset en het verzenden van elke updatebewerking per aanvraag naar de server een rekenboete oploopt.

De Bulk() API komt te hulp en verbetert de prestaties aanzienlijk, omdat schrijfbewerkingen slechts eenmaal in bulk naar de server worden verzonden. Efficiëntie wordt bereikt omdat de methode niet elk schrijfverzoek naar de server verzendt (zoals bij de huidige forEach() in de forEach() -lus) maar slechts één op de 1000 verzoeken, waardoor updates efficiënter en sneller zijn dan momenteel.

Met hetzelfde concept hierboven met de forEach() -lus om de batches te maken, kunnen we de verzameling als volgt in bulk bijwerken. In deze demonstratie gebruikt de Bulk() API beschikbaar in MongoDB-versies >= 2.6 en < 3.2 de initializeUnorderedBulkOp() -methode om de schrijfbewerkingen in de batches parallel en in een niet-deterministische volgorde uit te voeren.

Het werkt alle documenten in de klantenverzameling bij door de salary en dob velden te wijzigen in respectievelijk numerical en datetime waarden:

var bulk = db.test.initializeUnorderedBulkOp(),
    counter = 0; // counter to keep track of the batch update size

db.test.find({
    "salary": { "$exists": true, "$type": 2 },
    "dob": { "$exists": true, "$type": 2 }
}).snapshot().forEach(function(doc){ 
    var newSalary = parseInt(doc.salary),
        newDob = new ISODate(doc.dob);
    bulk.find({ "_id": doc._id }).updateOne({ 
        "$set": { "salary": newSalary, "dob": newDob }
    });

    counter++; // increment counter
    if (counter % 1000 == 0) {
        bulk.execute(); // Execute per 1000 operations and re-initialize every 1000 update statements
        bulk = db.test.initializeUnorderedBulkOp();
    }
});

Het volgende voorbeeld is van toepassing op de nieuwe MongoDB-versie 3.2 die sindsdien de API Bulk() heeft afgeschaft en een nieuwere set apis heeft geleverd met bulkWrite() .

Het gebruikt dezelfde cursors als hierboven, maar maakt de arrays met de bulkbewerkingen met dezelfde forEach() om elk forEach() naar de array te duwen. Omdat schrijfopdrachten niet meer dan 1000 bewerkingen kunnen accepteren, is het nodig om bewerkingen te groeperen om maximaal 1000 bewerkingen te hebben en de array opnieuw te initialiseren wanneer de lus de 1000 iteratie bereikt:

var cursor = db.test.find({
        "salary": { "$exists": true, "$type": 2 },
        "dob": { "$exists": true, "$type": 2 }
    }),
    bulkUpdateOps = [];

cursor.snapshot().forEach(function(doc){ 
    var newSalary = parseInt(doc.salary),
        newDob = new ISODate(doc.dob);
    bulkUpdateOps.push({ 
        "updateOne": {
            "filter": { "_id": doc._id },
            "update": { "$set": { "salary": newSalary, "dob": newDob } }
         }
    });

    if (bulkUpdateOps.length === 1000) {
        db.test.bulkWrite(bulkUpdateOps);
        bulkUpdateOps = [];
    }
});         

if (bulkUpdateOps.length > 0) { db.test.bulkWrite(bulkUpdateOps); }