64

I have a huge collection of documents in my DB and I'm wondering how can I run through all the documents and update them, each document with a different value.

3
  • 2
    It depends on the driver you're using to connect to MongoDB. Commented Aug 26, 2014 at 14:11
  • Im using the mongodb driver Commented Aug 26, 2014 at 14:15
  • Can you give me some example with the update inside the forEach(), and specify where are you closing the connection to the DB because i had problems with that Commented Aug 26, 2014 at 14:16

10 Answers 10

136

The answer depends on the driver you're using. All MongoDB drivers I know have cursor.forEach() implemented one way or another.

Here are some examples:

node-mongodb-native

collection.find(query).forEach(function(doc) {
  // handle
}, function(err) {
  // done or error
});

mongojs

db.collection.find(query).forEach(function(err, doc) {
  // handle
});

monk

collection.find(query, { stream: true })
  .each(function(doc){
    // handle doc
  })
  .error(function(err){
    // handle error
  })
  .success(function(){
    // final callback
  });

mongoose

collection.find(query).stream()
  .on('data', function(doc){
    // handle doc
  })
  .on('error', function(err){
    // handle error
  })
  .on('end', function(){
    // final callback
  });

Updating documents inside of .forEach callback

The only problem with updating documents inside of .forEach callback is that you have no idea when all documents are updated.

To solve this problem you should use some asynchronous control flow solution. Here are some options:

Here is an example of using async, using its queue feature:

var q = async.queue(function (doc, callback) {
  // code for your update
  collection.update({
    _id: doc._id
  }, {
    $set: {hi: 'there'}
  }, {
    w: 1
  }, callback);
}, Infinity);

var cursor = collection.find(query);
cursor.each(function(err, doc) {
  if (err) throw err;
  if (doc) q.push(doc); // dispatching doc to async.queue
});

q.drain = function() {
  if (cursor.isClosed()) {
    console.log('all items have been processed');
    db.close();
  }
}
Sign up to request clarification or add additional context in comments.

12 Comments

Can i use an update function inside that will get doc._id as a query from the for each?
Yes, you can. But you'll need to wait for all update operations to finish before closing your connection.
Can you add a Promise example to iterate over .forEach?
For what it worth, now the node-mongodb-native has an API called forEach that takes a callback.
Just a small note for mongoose - the .stream method is deprecated and now we should use .cursor
|
40

Using the mongodb driver, and modern NodeJS with async/await, a good solution is to use next():

const collection = db.collection('things')
const cursor = collection.find({
  bla: 42 // find all things where bla is 42
});
let document;
while ((document = await cursor.next())) {
  await collection.findOneAndUpdate({
    _id: document._id
  }, {
    $set: {
      blu: 43
    }
  });
}

This results in only one document at a time being required in memory, as opposed to e.g. the accepted answer, where many documents get sucked into memory, before processing of the documents starts. In cases of "huge collections" (as per the question) this may be important.

If documents are large, this can be improved further by using a projection, so that only those fields of documents that are required are fetched from the database.

9 Comments

@ZachSmith: this is correct, but very slow. It can be sped up 10X or more by using bulkWrite.
there is hasNext method. and it should be used instead of while(document=....) pattern
Why do you state this will only load 1 doc into memory? Don't you need to add to the find call tge batchSize option and assign it the value 1 for it to be that way?
@ZivGlazer that's not what I'm saying: The code presented requires only one doc in memory at any time. The batch size you are referring to determines how many documents get fetched in one request from the database, and that determines how many docs will reside in memory at any time, so you are right in that sense; but a batch size of 1 is probably not a good idea... finding a good batch size is a different question :)
I agree, I wanted to do a similar process of reading a big collection with big documents and updating each one I read, I end up implementing something like this:
|
21

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

MongoClient.connect('mongodb://localhost:27017/crunchbase', function(err, db) {

    assert.equal(err, null);
    console.log("Successfully connected to MongoDB.");

    var query = {
        "category_code": "biotech"
    };

    db.collection('companies').find(query).toArray(function(err, docs) {

        assert.equal(err, null);
        assert.notEqual(docs.length, 0);

        docs.forEach(function(doc) {
            console.log(doc.name + " is a " + doc.category_code + " company.");
        });

        db.close();

    });

});

Notice that the call .toArray is making the application to fetch the entire dataset.


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

MongoClient.connect('mongodb://localhost:27017/crunchbase', function(err, db) {

    assert.equal(err, null);
    console.log("Successfully connected to MongoDB.");

    var query = {
        "category_code": "biotech"
    };

    var cursor = db.collection('companies').find(query);

    function(doc) {
        cursor.forEach(
                console.log(doc.name + " is a " + doc.category_code + " company.");
            },
            function(err) {
                assert.equal(err, null);
                return db.close();
            }
    );
});

Notice that the cursor returned by the find() is assigned to var cursor. With this approach, instead of fetching all data in memory and consuming data at once, we're streaming the data to our application. find() can create a cursor immediately because it doesn't actually make a request to the database until we try to use some of the documents it will provide. The point of cursor is to describe our query. The 2nd parameter to cursor.forEach shows what to do when the driver gets exhausted or an error occurs.

In the initial version of the above code, it was toArray() which forced the database call. It meant we needed ALL the documents and wanted them to be in an array.

Also, MongoDB returns data in batch format. The image below shows, requests from cursors (from application) to MongoDB

MongoDB cursor requests

forEach is better than toArray because we can process documents as they come in until we reach the end. Contrast it with toArray - where we wait for ALL the documents to be retrieved and the entire array is built. This means we're not getting any advantage from the fact that the driver and the database system are working together to batch results to your application. Batching is meant to provide efficiency in terms of memory overhead and the execution time. Take advantage of it, if you can in your application.

1 Comment

cursor causes extra problems in handling data on client side.. the array is easier to consume due to the entire async nature of the system..
10

None of the previous answers mentions batching the updates. That makes them extremely slow 🐌 - tens or hundreds of times slower than a solution using bulkWrite.

Let's say you want to double the value of a field in each document. Here's how to do that fast 💨 and with fixed memory consumption:

// Double the value of the 'foo' field in all documents
let bulkWrites = [];
const bulkDocumentsSize = 100;  // how many documents to write at once
let i = 0;
db.collection.find({ ... }).forEach(doc => {
  i++;

  // Update the document...
  doc.foo = doc.foo * 2;

  // Add the update to an array of bulk operations to execute later
  bulkWrites.push({
    replaceOne: {
      filter: { _id: doc._id },
      replacement: doc,
    },
  });

  // Update the documents and log progress every `bulkDocumentsSize` documents
  if (i % bulkDocumentsSize === 0) {
    db.collection.bulkWrite(bulkWrites);
    bulkWrites = [];
    print(`Updated ${i} documents`);
  }
});
// Flush the last <100 bulk writes
db.collection.bulkWrite(bulkWrites);

Comments

5

And here is an example of using a Mongoose cursor async with promises:

new Promise(function (resolve, reject) {
  collection.find(query).cursor()
    .on('data', function(doc) {
      // ...
    })
    .on('error', reject)
    .on('end', resolve);
})
.then(function () {
  // ...
});

Reference:

3 Comments

If you have a sizable list of documents won't this exhaust memory? (ie: 10M docs)
@chovy in theory it shouldn't, this is why you use a cursor vs loading everything in an array in the first place. The promise then gets fulfilled only after the cursor ends or err. If you do have such a db then it shouldn't be too difficult to test this, I'd be curious myself.
you can use limit to avoid getting 10M docs, I don't know of any human who can read 10M docs at once, or a screen which can render them visually on one page (mobile, laptop or a desktop - forget about watches
5

You can now use (in an async function, of course):

for await (let doc of collection.find(query)) {
  await updateDoc(doc);
}

// all done

which nicely serializes all updates.

1 Comment

doc can be const because its scope is within the loop.
4

Leonid's answer is great, but I want to reinforce the importance of using async/promises and to give a different solution with a promises example.

The simplest solution to this problem is to loop forEach document and call an update. Usually, you don't need close the db connection after each request, but if you do need to close the connection, be careful. You must just close it if you are sure that all updates have finished executing.

A common mistake here is to call db.close() after all updates are dispatched without knowing if they have completed. If you do that, you'll get errors.

Wrong implementation:

collection.find(query).each(function(err, doc) {
  if (err) throw err;

  if (doc) {
    collection.update(query, update, function(err, updated) {
      // handle
    });
  } 
  else {
    db.close(); // if there is any pending update, it will throw an error there
  }
});

However, as db.close() is also an async operation (its signature have a callback option) you may be lucky and this code can finish without errors. It may work only when you need to update just a few docs in a small collection (so, don't try).


Correct solution:

As a solution with async was already proposed by Leonid, below follows a solution using Q promises.

var Q = require('q');
var client = require('mongodb').MongoClient;

var url = 'mongodb://localhost:27017/test';

client.connect(url, function(err, db) {
  if (err) throw err;

  var promises = [];
  var query = {}; // select all docs
  var collection = db.collection('demo');
  var cursor = collection.find(query);

  // read all docs
  cursor.each(function(err, doc) {
    if (err) throw err;

    if (doc) {

      // create a promise to update the doc
      var query = doc;
      var update = { $set: {hi: 'there'} };

      var promise = 
        Q.npost(collection, 'update', [query, update])
        .then(function(updated){ 
          console.log('Updated: ' + updated); 
        });

      promises.push(promise);
    } else {

      // close the connection after executing all promises
      Q.all(promises)
      .then(function() {
        if (cursor.isClosed()) {
          console.log('all items have been processed');
          db.close();
        }
      })
      .fail(console.error);
    }
  });
});

10 Comments

When I try this I get the error doc is not defined for the line var promises = calls.map(myUpdateFunction(doc));
@robocode, ty! I've fixed the error and the example is working fine now.
This approach requires you to load the whole dataset into memory, before processing the results. This isn't a practical approach for large datasets (you will run out of memory). You might as well just us .toArray(), which is much simpler (but also requires loading whole dataset of course).
@UpTheCreek, thanks for your feedback. I've updated my answer to store only the promise object instead of the doc object. It uses much less memory because the promises object is a small JSON that stores the state.
@UpTheCreek, for an even better solution regarding the memory usage, you would need to execute the query two times: first with a "count" and second to get the cursor. After each update is completed, you would increment a variable and only stops the program (and close the db connection) after this variable reach the total count of results.
|
4

The node-mongodb-native now supports a endCallback parameter to cursor.forEach as for one to handle the event AFTER the whole iteration, refer to the official document for details http://mongodb.github.io/node-mongodb-native/2.2/api/Cursor.html#forEach.

Also note that .each is deprecated in the nodejs native driver now.

2 Comments

Can you provide an example of firing a callback after all results have been processed?
@chovy in forEach(iteratorCallback, endCallback) endCallback(error) is called when there is no more data (with error is undefined).
1

let's assume that we have the below MongoDB data in place.

Database name: users
Collection name: jobs
===========================
Documents
{ "_id" : ObjectId("1"), "job" : "Security", "name" : "Jack", "age" : 35 }
{ "_id" : ObjectId("2"), "job" : "Development", "name" : "Tito" }
{ "_id" : ObjectId("3"), "job" : "Design", "name" : "Ben", "age" : 45}
{ "_id" : ObjectId("4"), "job" : "Programming", "name" : "John", "age" : 25 }
{ "_id" : ObjectId("5"), "job" : "IT", "name" : "ricko", "age" : 45 }
==========================

This code:

var MongoClient = require('mongodb').MongoClient;
var dbURL = 'mongodb://localhost/users';

MongoClient.connect(dbURL, (err, db) => {
    if (err) {
        throw err;
    } else {
        console.log('Connection successful');
        var dataBase = db.db();
        // loop forEach
        dataBase.collection('jobs').find().forEach(function(myDoc){
        console.log('There is a job called :'+ myDoc.job +'in Database')})
});

Comments

0

I looked for a solution with good performance and I end up creating a mix of what I found which I think works good:

/**
 * This method will read the documents from the cursor in batches and invoke the callback
 * for each batch in parallel.
 * IT IS VERY RECOMMENDED TO CREATE THE CURSOR TO AN OPTION OF BATCH SIZE THAT WILL MATCH
 * THE VALUE OF batchSize. This way the performance benefits are maxed out since
 * the mongo instance will send into our process memory the same number of documents
 * that we handle in concurrent each time, so no memory space is wasted
 * and also the memory usage is limited.
 *
 * Example of usage:
 * const cursor = await collection.aggregate([
     {...}, ...],
     {
        cursor: {batchSize: BATCH_SIZE} // Limiting memory use
    });
 DbUtil.concurrentCursorBatchProcessing(cursor, BATCH_SIZE, async (doc) => ...)
 * @param cursor - A cursor to batch process on.
 * We can get this from our collection.js API by either using aggregateCursor/findCursor
 * @param batchSize - The batch size, should match the batchSize of the cursor option.
 * @param callback - Callback that should be async, will be called in parallel for each batch.
 * @return {Promise<void>}
 */
static async concurrentCursorBatchProcessing(cursor, batchSize, callback) {
    let doc;
    const docsBatch = [];

    while ((doc = await cursor.next())) {
        docsBatch.push(doc);

        if (docsBatch.length >= batchSize) {
            await PromiseUtils.concurrentPromiseAll(docsBatch, async (currDoc) => {
                return callback(currDoc);
            });

            // Emptying the batch array
            docsBatch.splice(0, docsBatch.length);
        }
    }

    // Checking if there is a last batch remaining since it was small than batchSize
    if (docsBatch.length > 0) {
        await PromiseUtils.concurrentPromiseAll(docsBatch, async (currDoc) => {
            return callback(currDoc);
        });
    }
}

An example of usage for reading many big documents and updating them:

        const cursor = await collection.aggregate([
        {
            ...
        }
    ], {
        cursor: {batchSize: BATCH_SIZE}, // Limiting memory use 
        allowDiskUse: true
    });

    const bulkUpdates = [];

    await DbUtil.concurrentCursorBatchProcessing(cursor, BATCH_SIZE, async (doc: any) => {
        const update: any = {
            updateOne: {
                filter: {
                    ...
                },
                update: {
                   ...
                }
            }
        };            

        bulkUpdates.push(update);

        // Updating if we read too many docs to clear space in memory
        await this.bulkWriteIfNeeded(bulkUpdates, collection);
    });

    // Making sure we updated everything
    await this.bulkWriteIfNeeded(bulkUpdates, collection, true);

...

    private async bulkWriteParametersIfNeeded(
    bulkUpdates: any[], collection: any,
    forceUpdate = false, flushBatchSize) {

    if (bulkUpdates.length >= flushBatchSize || forceUpdate) {
        // concurrentPromiseChunked is a method that loops over an array in a concurrent way using lodash.chunk and Promise.map
        await PromiseUtils.concurrentPromiseChunked(bulkUpsertParameters, (upsertChunk: any) => {
            return techniquesParametersCollection.bulkWrite(upsertChunk);
        });

        // Emptying the array
        bulkUpsertParameters.splice(0, bulkUpsertParameters.length);
    }
}

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.