MongoDB
Bulkoperationer

Anmärkningar

Konvertera ett fält till en annan typ och uppdatera hela samlingen i Bulk

Vanligtvis fallet när man vill ändra en fälttyp till en annan, till exempel den ursprungliga samlingen kan ha "numeriska" eller "datum" fält sparade som strängar:

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

Målet skulle vara att uppdatera en humongös samling som ovan till

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

För relativt små data kan man uppnå ovanstående genom att iterera samlingen med hjälp av en snapshot med markörens forEach() -metod och uppdatera varje dokument på följande sätt:

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

Även om detta är optimalt för små samlingar, är prestanda med stora samlingar kraftigt minskat eftersom looping genom ett stort datasätt och skickning av varje uppdateringsoperation per begäran till servern medför en beräkningstraff.

Bulk() API kommer till undsättning och förbättrar kraftigt prestandan eftersom skrivoperationer skickas till servern endast en gång i bulk. Effektivitet uppnås eftersom metoden inte skickar varje skrivbegäran till servern (som med det aktuella uppdateringsförklaringen inom forEach() ) utan bara en gång i 1000 förfrågningar, vilket gör uppdateringar mer effektiva och snabbare än nu är.

Genom att använda samma koncept ovan med forEach() att skapa partierna, kan vi uppdatera samlingen i bulk på följande sätt. I denna demonstration använder Bulk() API som är tillgängligt i MongoDB-versioner >= 2.6 och < 3.2 metoden initializeUnorderedBulkOp() för att utföra parallellt, såväl som i en icke-bestämd ordning, skrivoperationerna i batcharna.

Den uppdaterar alla dokument i klienter samlingen genom att ändra salary och dob fälten till numerical och datetime värden respektive:

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

Nästa exempel gäller den nya MongoDB version 3.2 som sedan fasas ut i Bulk() API och gav en nyare uppsättning API: er med hjälp av bulkWrite() .

Den använder samma markörer som ovan men skapar matriserna med bulkoperationerna med samma forEach() för att skjuta varje bulkskrivningsdokument till arrayen. Eftersom skrivkommandon inte kan acceptera högst 1000 operationer, finns det behov av att gruppera operationer för att ha högst 1000 operationer och återintalisera matrisen när slingan träffar 1000-iterationen:

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