如何最高效地更新MongoDB中的大量文档？

我想要最有效地更新大量 (> 100,000) 文档。

我的第一个天真的方法是在 JS 级别上进行，编写脚本 首先获取 _ids，然后循环 _ids 并通过 _id 调用更新（完整 文档或 $set 补丁）。

我遇到了内存问题，还将数据分成了最大块。 500 文档（打开和关闭连接）似乎效果不佳。

那么我该如何在 MongoDB 级别解决这个问题呢？
最佳实践？

我有 3 个常见用例，通常是维护工作流程：

1. 更改属性值的类型，但不更改值。

// before
{
  timestamp : '1446987395'
}

// after
{
  timestamp : 1446987395
}

2. 根据现有财产的价值添加新财产。

// before
{
  firstname : 'John',
  lastname  : 'Doe'
}

// after
{
  firstname : 'John',
  lastname  : 'Doe',
  name      : 'John Doe'
}

3. 只需添加从文档中删除的属性。

// before
{
  street    : 'Whatever Ave',
  street_no : '1025'
}

// after
{
  street    : 'Whatever Ave',
  no        : '1025'
}

感谢您的帮助。

如果您的 MongoDB 服务器是 2.6 或更新版本，最好利用写入命令Bulk API http://blog.mongodb.org/post/84922794768/mongodbs-new-bulk-api允许批量执行update https://docs.mongodb.org/v3.0/reference/method/Bulk.find.updateOne/#Bulk.find.updateOne操作只是服务器顶部的抽象，可以轻松构建批量操作。这些批量操作主要有两种类型：

• 订单批量操作。这些操作按顺序执行所有操作，并在第一个写入错误时出错。
• 无序批量操作。这些操作并行执行所有操作并汇总所有错误。无序批量操作不保证执行顺序。

请注意，对于 2.6 之前的旧服务器，API 将对操作进行下转换。然而，不可能 100% 下变频，因此可能存在一些边缘情况，无法正确报告正确的数字。

对于您的三个常见用例，您可以像这样实现批量 API：

Case 1.更改属性值的类型，而不更改值：

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

Case 2.根据现有财产的价值添加新财产：

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

Case 3.只需添加从文档中删除属性即可。

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