2

For a mobile application which will hold quite some documents, we've chosen Mongo as a database.

Let me explain the situation first:

Every day, I have a set of documents that needs to be inserted, so I drop all the documents and insert them again - in total there are 200.000 documents which are quite fast imported using a bulk import.

After than file, I do have a CSV file with 500.000 documents and for each entry I must query the database to see if a document with a given value out of the current CSV line does exist, if so, I need to update it, otherwise, continue to the next.

This operation is currently taken 11 hours, which is way too long. Therefore, I wanted to go to a sharded environment.

I have a HOST server which is running vSphere and on top of that, I've created 3 configuration servers, 1 router server and 6 database servers.

Now, I setup sharding using a hashed value of the '_id' field as the sharding key (my database collection is empty) and I execute a script which imports 10 times 10.000 documents, thus 100.000 documents in total and provide me the average speed.

The following commands are all executed from the Mongo Router Server.

  • Add a single Shard to the database:

    mongos> sh.addShard("10.20.2.85:27017")
    
  • Enable sharding on database level:

    mongos> sh.enableSharding("testData")
    
  • Shard a collection based on a hashed '_id' value (collection exists and a hashed index for the '_id' field is created).

    mongos> sh.shardCollection("testData.testData" , { "_id": "hashed" } )
    

When I request the sharding status sh.status(), I do receive the following output:

--- Sharding Status ---
  sharding version: {
        "_id" : 1,
        "minCompatibleVersion" : 5,
        "currentVersion" : 6,
        "clusterId" : ObjectId("56c42f5d2cdddd79fb34411f")
}
  shards:
        {  "_id" : "shard0000",  "host" : "10.20.2.85:27017" }
  active mongoses:
        "3.2.2" : 1
  balancer:
        Currently enabled:  yes
        Currently running:  no
        Failed balancer rounds in last 5 attempts:  0
        Migration Results for the last 24 hours:
                14 : Success
  databases:
        {  "_id" : "testData",  "primary" : "shard0000",  "partitioned" :     true }


            testData.testData
                    shard key: { "_id" : "hashed" }
                    unique: false
                    balancing: true
                    chunks:
                            shard0000       1
                    { "_id" : { "$minKey" : 1 } } -->> { "_id" : { "$maxKey" : 1 } } on : shard0000 Timestamp(1, 0)

I do have a custom script to insert the 10 times 10.000 documents (see the end of this post for the script). When I do run this script, the average throughput (documents per seconds) is 2229.76 documents / seconds.

Ok, I do restart right now and I will use 6 shards.

  • Add a single Shard to the database:

    mongos> sh.addShard("10.20.2.85:27017")
    mongos> sh.addShard("10.20.2.86:27017")
    mongos> sh.addShard("10.20.2.87:27017")
    mongos> sh.addShard("10.20.2.88:27017")
    mongos> sh.addShard("10.20.2.89:27017")
    mongos> sh.addShard("10.20.2.90:27017")
    
  • Enable sharding on database level:

    mongos> sh.enableSharding("testData")
    
  • Shard a collection based on a hashed '_id' value (collection exists and a hashed index for the '_id' field is created).

    mongos> sh.shardCollection("testData.testData" , { "_id": "hashed" } )
    

When I request the sharding status sh.status(), I do receive the following output:

--- Sharding Status ---
  sharding version: {
        "_id" : 1,
        "minCompatibleVersion" : 5,
        "currentVersion" : 6,
        "clusterId" : ObjectId("56c42f5d2cdddd79fb34411f")
}
  shards:
        {  "_id" : "shard0000",  "host" : "10.20.2.85:27017" }
        {  "_id" : "shard0001",  "host" : "10.20.2.86:27017" }
        {  "_id" : "shard0002",  "host" : "10.20.2.87:27017" }
        {  "_id" : "shard0003",  "host" : "10.20.2.88:27017" }
        {  "_id" : "shard0004",  "host" : "10.20.2.89:27017" }
        {  "_id" : "shard0005",  "host" : "10.20.2.90:27017" }
  active mongoses:
        "3.2.2" : 1
  balancer:
        Currently enabled:  yes
        Currently running:  no
        Failed balancer rounds in last 5 attempts:  0
        Migration Results for the last 24 hours:
                19 : Success
  databases:
        {  "_id" : "testData",  "primary" : "shard0000",  "partitioned" :     true }
                testData.testData
                        shard key: { "_id" : "hashed" }
                        unique: false
                        balancing: true
                        chunks:
                                shard0000       2
                                shard0001       2
                                shard0002       2
                                shard0003       2
                                shard0004       2
                                shard0005       2
                        { "_id" : { "$minKey" : 1 } } -->> { "_id" :     NumberLong("-7686143364045646500") } on : shard0000 Timestamp(6, 2)
                        { "_id" : NumberLong("-7686143364045646500") } -->>     { "_id" : NumberLong("-6148914691236517200") } on : shard0000 Timestamp(6, 3)
                        { "_id" : NumberLong("-6148914691236517200") } -->>     { "_id" : NumberLong("-4611686018427387900") } on : shard0001 Timestamp(6, 4)
                        { "_id" : NumberLong("-4611686018427387900") } -->>     { "_id" : NumberLong("-3074457345618258600") } on : shard0001 Timestamp(6, 5)
                        { "_id" : NumberLong("-3074457345618258600") } -->>     { "_id" : NumberLong("-1537228672809129300") } on : shard0002 Timestamp(6, 6)
                        { "_id" : NumberLong("-1537228672809129300") } -->>     { "_id" : NumberLong(0) } on : shard0002 Timestamp(6, 7)
                        { "_id" : NumberLong(0) } -->> { "_id" :     NumberLong("1537228672809129300") } on : shard0003 Timestamp(6, 8)
                        { "_id" : NumberLong("1537228672809129300") } -->> {     "_id" : NumberLong("3074457345618258600") } on : shard0003 Timestamp(6, 9)
                        { "_id" : NumberLong("3074457345618258600") } -->> {     "_id" : NumberLong("4611686018427387900") } on : shard0004 Timestamp(6, 10)
                        { "_id" : NumberLong("4611686018427387900") } -->> {     "_id" : NumberLong("6148914691236517200") } on : shard0004 Timestamp(6, 11)
                        { "_id" : NumberLong("6148914691236517200") } -->> {     "_id" : NumberLong("7686143364045646500") } on : shard0005 Timestamp(6, 12)
                        { "_id" : NumberLong("7686143364045646500") } -->> {     "_id" : { "$maxKey" : 1 } } on : shard0005 Timestamp(6, 13)

When I do run the script again to insert the 10 times 10.000 documents I get an average insert speed of 2162.39, this does mean that adding shards does not increase the insert performance, in fact the insert performance is lower when adding shards.

There should be 10.000 documents in my collection right now, because the collection is always cleared).

Each shard does contain roughly 1600 documents which means the sharding key is good because the documents are evenly distributed.

This is really driving me nuts. I tought that sharding was used to get a better performance but instead I do get a lower performance.

Can anyone help me find the light to make sharding work correctly?

For the persons that are interested, please find the insert script below:

// Set the variables that defines how many times the test is being executed.
var amountOfDocumentsToInsert = 10000;
var amountOfRuns = 10;

// Get the sum of all the elements in an array.
function getArraySum(source) {
  var total = 0;

  // Loop over all the elements in the array and add them up.
  for( var i = 0 ; i < source.length ; i++ ){
    total += parseFloat (source [i ]);
      }

  return total.toFixed (2);
}

// Get the average of all the items in an array.
function getArrayAverage(source) {
  var average = getArraySum(source ) / source.length ;

  return average.toFixed (2);
}

// Defines a function that will insert the given amount of documents.
function insertDocuments(database, amount ) {

  // Remove all documents in the collection.
  db.testData.remove({});

  // Defines the start time of the function.
  var startTime = Date.now();

  // Get the current amount of documents in the collection (this should be 0).
  var originalCount = db.testData.find ({}).count ();

  // Put the application in a loop and create 10.000 documents.
  for ( var x = 0 ; x < amount; x ++ ) {
    db.testData.insert({
      time : Date.now (),
      index : Math.floor (Math.random () * 10000 ) + 1
    });
  }

  // Defines the end time of the function.
  var endTime = Date.now();

  // Get the total running time.
  var runningTime = (endTime - startTime );

  // Get the amount of documents that are inserted.
  var insertedDocuments = db.testData.find ({}).count () - originalCount;

  // Get the throughput of the application, which are the amount of documents that are inserted per seconds.
  var throughput = (insertedDocuments / runningTime * 1000);

  // Returns the object.
  return {
       runningTime : runningTime ,
       documentsCount : insertedDocuments ,
       throughput : throughput
  };
}

// Ensure that 'db' does reference the database with name 'testData'.
db = db.getSiblingDB('testData' );

var diagnostics = "";
var documentsCount = [];
var runningTime = [];
var throughput = [];

// Run the scenario a couple of times.
for (var i = 0; i < amountOfRuns; i++) {
  var insertRun = insertDocuments(db, amountOfDocumentsToInsert);

  // Creates the message to print.
  diagnostics = diagnostics + "[RUN " + (i + 1) + "]: Inserted Documents: " + insertRun.documentsCount + ".\r\n" +
                              "[RUN " + (i + 1) + "]: Required Time: " + insertRun.runningTime + " milliseconds.\r\n" +
                              "[RUN " + (i + 1) + "]: Throughput: " + insertRun.throughput + " documents / seconds.\r\n" +
                              "-- COMPLETED --\r\n";

  // Push the items on an array to that averages can be read.
  documentsCount.push(insertRun.documentsCount);
  runningTime.push(insertRun.runningTime);
  throughput.push(insertRun.throughput);
}

// Print some diagnostics.
diagnostics = diagnostics + "[TOTAL]: Inserted Documents: " + getArraySum(documentsCount) + "\r\n" +
                            "[AVERAGE]: Required Time: " + getArrayAverage(runningTime) + " milliseconds.\r\n" +
                            "[AVERAGE]: Throughput: " + getArrayAverage(throughput) + " documents / seconds. \r\n";

// Print some diagnostics to the console.
print(diagnostics );

Kind regards

  • Can you retry the load test using numInitialChunks of 600 (100 for each shard)?db.runCommand( { shardCollection: "testData.testData", key: { "_id": "hashed" }, numInitialChunks : 600 } ) – Antonios Feb 21 '16 at 0:51

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.