Làm thế nào để cập nhật một số lượng lớn tài liệu trong MongoDB một cách hiệu quả nhất?

Nếu máy chủ MongoDB của bạn là 2.6 hoặc mới hơn, tốt hơn nên tận dụng lợi thế của việc sử dụng lệnh ghi API hàng loạt cho phép thực thi số lượng lớn update các thao tác chỉ đơn giản là các thao tác trừu tượng trên máy chủ để dễ dàng xây dựng các hoạt động hàng loạt. Các hoạt động hàng loạt này chủ yếu có hai loại:

• Hoạt động hàng loạt có thứ tự . Các thao tác này thực hiện tất cả các thao tác theo thứ tự và không xảy ra lỗi trong lỗi ghi đầu tiên.
• Hoạt động hàng loạt không có thứ tự . Các hoạt động này thực hiện tất cả các hoạt động song song và tổng hợp tất cả các lỗi. Các hoạt động hàng loạt không theo thứ tự không đảm bảo thứ tự thực hiện.

Lưu ý, đối với các máy chủ cũ hơn 2.6, API sẽ chuyển đổi các hoạt động xuống. Tuy nhiên, không thể downconvert 100% vì vậy có thể có một số trường hợp phức tạp trong đó nó không thể báo cáo chính xác các con số phù hợp.

Đối với ba trường hợp sử dụng phổ biến của bạn, bạn có thể triển khai API Bulk như sau:

Trường hợp 1. Thay đổi loại giá trị của thuộc tính mà không thay đổi giá trị:

var MongoClient = require('mongodb').MongoClient;

MongoClient.connect("mongodb://localhost:27017/test", function(err, db) {
    // Handle error
    if(err) throw err;

    // Get the collection and bulk api artefacts
    var col = db.collection('users'),           
        bulk = col.initializeOrderedBulkOp(), // Initialize the Ordered Batch
        counter = 0;        

    // Case 1. Change type of value of property, without changing the value.        
    col.find({"timestamp": {"$exists": true, "$type": 2} }).each(function (err, doc) {

        var newTimestamp = parseInt(doc.timestamp);
        bulk.find({ "_id": doc._id }).updateOne({
            "$set": { "timestamp": newTimestamp }
        });

        counter++;

        if (counter % 1000 == 0 ) {
            bulk.execute(function(err, result) {  
                // re-initialise batch operation           
                bulk = col.initializeOrderedBulkOp();
            });
        }
    });

    if (counter % 1000 != 0 ){
        bulk.execute(function(err, result) {
            // do something with result
            db.close();
        }); 
    } 
});

Trường hợp 2. Thêm thuộc tính mới dựa trên giá trị của thuộc tính hiện có:

MongoClient.connect("mongodb://localhost:27017/test", function(err, db) {
    // Handle error
    if(err) throw err;

    // Get the collection and bulk api artefacts
    var col = db.collection('users'),           
        bulk = col.initializeOrderedBulkOp(), // Initialize the Ordered Batch
        counter = 0;        

    // Case 2. Add new property based on value of existing property.        
    col.find({"name": {"$exists": false } }).each(function (err, doc) {

        var fullName = doc.firstname + " " doc.lastname;
        bulk.find({ "_id": doc._id }).updateOne({
            "$set": { "name": fullName }
        });

        counter++;

        if (counter % 1000 == 0 ) {
            bulk.execute(function(err, result) {  
                // re-initialise batch operation           
                bulk = col.initializeOrderedBulkOp();
            });
        }
    });

    if (counter % 1000 != 0 ){
        bulk.execute(function(err, result) {
            // do something with result
            db.close();
        }); 
    } 
});

Trường hợp 3. Chỉ cần thêm loại bỏ thuộc tính khỏi tài liệu.

MongoClient.connect("mongodb://localhost:27017/test", function(err, db) {
    // Handle error
    if(err) throw err;

    // Get the collection and bulk api artefacts
    var col = db.collection('users'),           
        bulk = col.initializeOrderedBulkOp(), // Initialize the Ordered Batch
        counter = 0;        

    // Case 3. Simply adding removing properties from documents.    
    col.find({"street_no": {"$exists": true } }).each(function (err, doc) {

        bulk.find({ "_id": doc._id }).updateOne({
            "$set": { "no": doc.street_no },
            "$unset": { "street_no": "" }
        });

        counter++;

        if (counter % 1000 == 0 ) {
            bulk.execute(function(err, result) {  
                // re-initialise batch operation           
                bulk = col.initializeOrderedBulkOp();
            });
        }
    });

    if (counter % 1000 != 0 ){
        bulk.execute(function(err, result) {
            // do something with result
            db.close();
        }); 
    } 
});