MongoDB
Массовые операции

замечания

Преобразование поля в другой тип и обновление всей коллекции в Bulk

Обычно случай, когда вы хотите изменить тип поля на другой, например, исходная коллекция может иметь «числовые» или «даты» поля, сохраненные как строки:

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

Целью было бы обновить обширную коллекцию, как показано выше

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

Для относительно небольших данных можно добиться вышеуказанного путем итерации коллекции с помощью snapshot с помощью метода курсора forEach() и обновления каждого документа следующим образом:

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

Несмотря на то, что это оптимально для небольших коллекций, производительность с большими коллекциями значительно снижается, так как цикл через большой набор данных и отправка каждой операции обновления для каждого запроса на сервер приводит к компьютерному штрафу.

API Bulk() приходит на помощь и значительно повышает производительность, поскольку операции записи отправляются на сервер только один раз в массовом порядке. Эффективность достигается, поскольку метод не отправляет каждый запрос на запись на сервер (как и в текущем операторе обновления в цикле forEach() ), а только один раз в каждые 1000 запросов, что делает обновления более эффективными и быстрыми, чем в настоящее время.

Используя ту же концепцию выше, что и для цикла forEach() для создания пакетов, мы можем обновить коллекцию навалом, как показано ниже. В этой демонстрации API Bulk() доступный в версиях MongoDB >= 2.6 и < 3.2 использует метод initializeUnorderedBulkOp() для выполнения параллельно, а также в недетерминированном порядке операций записи в партиях.

Он обновляет все документы в коллекции клиентов, изменяя поля salary и dob на numerical и значения datetime и datetime соответственно:

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

Следующий пример относится к новой версии MongoDB 3.2 которая с тех пор устарела от API Bulk() и предоставила новый набор apis, используя bulkWrite() .

Он использует те же курсоры, что и выше, но создает массивы с объемными операциями, используя один и тот же метод курсора forEach() чтобы выталкивать каждый массивный документ записи в массив. Поскольку команды записи могут принимать не более 1000 операций, необходимо сгруппировать операции, чтобы иметь не более 1000 операций и повторно инициализировать массив, когда цикл достигает 1000 итераций:

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