MongoDB sharded cluster chunks distribution

dockermongodbmongodb-3.6sharding

I have deployed MongoDB cluster with two shards (two replica sets) with docker-compose.

Unfortunately I'm having problems with chunks distribution between shards, I'm new to MongoDB and database sin general so those problems can be result of me misunderstanding the topic. The questions describing problems are at the bottom of post, below is configuration of my sharded cluster.

Sharded cluster

It consists of:

Config replica set, each of three mongod instances configured with: mongod –configsvr –replSet configReplicaSet –bind_ip 0.0.0.0 –dbpath /data/db –port 27019
First replica shard of three mongod instances, each configured with: mongod –shardsvr –replSet shardReplicaSet1 –bind_ip 0.0.0.0 –dbpath /data/db –port 27017
Second replica shard of three mongod instances, each configured with: mongod –shardsvr –replSet shardReplicaSet2 –bind_ip 0.0.0.0 –dbpath /data/db –port 27017
Single mongos instance, it's container starts with /bin/bash despite command in docker-compose, it's duo to fact that the entrypoint defined in docker image is /bin/bash so it probably overrides docker-compose
Single mongo shell instance.

Here is docker-compose.yaml:

version: '2.1'
services:
  mongors1-1:
    container_name: mongors1-1
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet1 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27011:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.11
    stdin_open: true
    tty: true

  mongors1-2:
    container_name: mongors1-2
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet1 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27012:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.12
    stdin_open: true
    tty: true

  mongors1-3:
    container_name: mongors1-3
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet1 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27013:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.13
    stdin_open: true
    tty: true

  mongors2-1:
    container_name: mongors2-1
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet2 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27021:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.21
    stdin_open: true
    tty: true

  mongors2-2:
    container_name: mongors2-2
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet2 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27022:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.22
    stdin_open: true
    tty: true

  mongors2-3:
    container_name: mongors2-3
    image: mongodb-server-shard
    command: mongod --shardsvr --replSet shardReplicaSet2 --bind_ip 0.0.0.0  --dbpath /data/db --port 27017
    ports:
    - 27023:27017
    expose:
    - "27017"
    networks:
      default:
        ipv4_address: 172.19.0.23
    stdin_open: true 
    tty: true

  mongocfg1:
    container_name: mongocfg1
    image: mongodb-server-shard
    command: mongod --configsvr --replSet configReplicaSet --bind_ip 0.0.0.0  --dbpath /data/db --port 27019
    ports:
    - 2001:27019
    expose:
    - "27019"
    networks:
      default:
        ipv4_address: 172.19.0.101
    stdin_open: true
    tty: true

  mongocfg2:
    container_name: mongocfg2
    image: mongodb-server-shard
    command: mongod --configsvr --replSet configReplicaSet --bind_ip 0.0.0.0 --dbpath /data/db --port 27019
    ports:
    - 2002:27019
    expose:
    - "27019"
    networks:
      default:
        ipv4_address: 172.19.0.102
    stdin_open: true
    tty: true

  mongocfg3:
    container_name: mongocfg3
    image: mongodb-server-shard
    command: mongod --configsvr --replSet configReplicaSet --bind_ip 0.0.0.0 --dbpath /data/db --port 27019
    ports:
    - 2003:27019
    expose:
    - "27019"
    networks:
      default:
        ipv4_address: 172.19.0.103
    stdin_open: true 
    tty: true

  mongo-client:
    container_name: mongo-client1
    image: mongodb-client
    ports:
    - 5000:5000
    expose:
    - "5000"
    networks:
     default:
        ipv4_address: 172.19.0.50
    stdin_open: true 
    tty: true

  mongos1:
    container_name: mongos1
    image: mongodb-mongos
    command: mongos --configdb configReplicaSet/mongocfg1:27019,mongocfg2:27019 --bind_ip 0.0.0.0
    ports:
    - 3300:3300
    expose:
    - "3300"
    - "27017"
    networks:
     default:
        ipv4_address: 172.19.0.150
    stdin_open: true 
    tty: true

  # iperf3-server:
  #   container_name: iperf3-server
  #   image: networkstatic/iperf3
  #   expose:
  #   - "5201"
  #   command: iperf -s
  #   networks:
  #    default:
  #       ipv4_address: 172.19.0.200
  #   stdin_open: true 
  #   tty: true

  # iperf3-client:
  #   container_name: iperf3-client
  #   image: networkstatic/iperf3
  #   expose:
  #   - "5202"
  #   command: iperf -c 172.19.0.200
  #   networks:
  #    default:
  #       ipv4_address: 172.19.0.201
  #   stdin_open: true 
  #   tty: true

I'm not using official docker hub images but my own, they are very simple, basic installation, the difference is that containers with mongod for example have only mongod installed instead of whole MongoDB installation, similarly, mongos is only mongos installed. I think that this might be source of problem.

iperf3 containers were used for measuring speed of Docker bridged network, they resulted with over 50 Gbites/s of bandwidth, so it should not be the source of problems.

The further configuration of cluster is following:

I connect from mongo-client1 (after using docker attach) to one of config servers and run

rs.initiate(
{
_id: "configReplicaSet",
configsvr: true,
members: [
{ _id : 0, host : "mongocfg1:27019" },
{ _id : 1, host : "mongocfg2:27019" },
{ _id : 2, host : "mongocfg3:27019" }
]
}
)
Similarly I connect to one of shardReplicaSet1 mongod instances and run:

rs.initiate(
{
_id : "shardReplicaSet1",
members: [
{ _id : 0, host : "mongors1-1:27017" },
{ _id : 1, host : "mongors1-2:27017" },
{ _id : 2, host : "mongors1-3:27017"}
]})
In similar way I initate shardReplicaSet2.
I use docker attach to mongos1 container and start mongos with: mongos –configdb configReplicaSet/mongocfg1:27019,mongocfg2:27019 –bind_ip 0.0.0.0
1. I connect to mongos1 from mongo-client1 and run: sh.addShard( "shardReplicaSet1/mongors1-1:27017") and sh.addShard( "shardReplicaSet2/mongors2-1:27017"), I don't know if this is necessary but I use hostname of primary member of each replica set.

The problems

Being connected to mongos1, I create database named "key_v" with collection named "performance", then I
enable sharding on database (sh.enableSharding("key_v")),
then on collection using shard key:
sh.shardCollection("key_v.performance", {"address": 1})

Then I use for loop to insert many documents to it:

for(i=0; i<=10000; i++)
{
db.collection.insert("address": i,"array1": [999, "foo"]);
}

Oh, and I set chunk size to 1 MB
previously. So, after loop finishes, the results of
db.collection.getShardDistribution() shows that all chunks are located
on only one of shards, after few minutes in fact chunks are
distributed evenly between shards. Here is my question: should not
mongos distribute chunks between shards during execution of loop,
instead of directing them all to one shard? If it can configured,
how should I achieve that? By this?
[https://docs.mongodb.com/manual/tutorial/manage-shard-zone/][1] . Maybe shard key is incorrect?

Another related case: I try to import large MongoDB database in bson using mongorestore. I'm doing it from outside of Docker network:
mongorestore –host 127.19.0.150:3300 -d import1 -c test /path/base.bson
The import works well, but all the chunks are located in one of shards. Furthermore, migration of this chunks to second shard is very slow, the database weighs 1.3 Gbytes and after 40 minutes only 32% of chunks have been migrated. The database data schema consists of multiple fields, I have chosen one field with Int32 datatype as shard key, but it's cardinality is very low, 15% of documents have the same value for it, could this be a source of it? Estimated data per chunk is 529KiB (getShardDistribution() method results).

All containers run on computer with 32GB of RAM and i7-6700HQ, could the slow HDD be a bottleneck resulting in such a slow chunks migration?

Best Answer

So, after loop finishes, the results of db.collection.getShardDistribution() shows that all chunks are located on only one of shards, after few minutes in fact chunks are distributed evenly between shards. Here is my question: should not mongos distribute chunks between shards during execution of loop, instead of directing them all to one shard?

The issue is that you are using a monotonically increasing shard key. This will result in all inserts targeting the single shard (aka a "hot shard") that currently has the chunk range representing the highest shard key value. Data will eventually be rebalanced to other shards, but this is not an effective use of sharding for write scaling. You need to choose a more appropriate shard key. If your use case does not require range queries, you could also consider using a hashed shard key on the address field. See Hashed vs Ranged Sharding for an illustration of the expected outcomes.

Another related case: I try to import large MongoDB database in bson using mongorestore. I'm doing it from outside of Docker network: mongorestore --host 127.19.0.150:3300 -d import1 -c test /path/base.bson The import works well, but all the chunks are located in one of shards.

If the outcome is similar (all inserts going to a single shard and then being rebalanced), this also suggests a poor shard key choice.

If you are bulk inserting into an empty sharded collection there is an approach you can use to minimize rebalancing: pre-splitting chunk ranges based on the known distribution of shard key values in your existing data.

The database data schema consists of multiple fields, I have chosen one field with Int32 datatype as shard key, but it's cardinality is very low, 15% of documents have the same value for it, could this be a source of it?

Low cardinality shard keys will definitely cause issues with data distribution. The most granular chunk range possible will represent a single shard key value. If a large percentage of your documents share the same shard key values, those will eventually lead to indivisible jumbo chunks which will be ignored by the balancer.

All containers run on computer with 32GB of RAM and i7-6700HQ, could the slow HDD be a bottleneck resulting in such a slow chunks migration?

There isn't enough information to determine if your disk is the most limiting factor, but running a sharded cluster on a single computer with a slow HDD will certainly add resource contention challenges. Choosing appropriate shard keys should minimize the need for data migration unless you are adding or removing shards for your deployment.

Assuming you are using a recent version of MongoDB with the WiredTiger storage engine as default (MongoDB 3.2+), you will definitely want to explicitly set --wiredTigerCacheSizeGB to limit the internal cache size for mongod instances. See: To what size should I set the WiredTiger internal cache?.

Related Solutions

Mongodb – How to Determine Chunk Distribution (data and number of docs) in a Sharded MongoDB Cluster

There is currently no built-in way to do this, so a small function is needed. For the purposes of this answer I have created a 2 shard cluster with ~1 million documents as per these instructions. Next up I used this function to examine those documents:

AllChunkInfo = function(ns, est){
    var chunks = db.getSiblingDB("config").chunks.find({"ns" : ns}).sort({min:1}); //this will return all chunks for the ns ordered by min
    //some counters for overall stats at the end
    var totalChunks = 0;
    var totalSize = 0;
    var totalEmpty = 0;
    print("ChunkID,Shard,ChunkSize,ObjectsInChunk"); // header row
    // iterate over all the chunks, print out info for each 
    chunks.forEach( 
        function printChunkInfo(chunk) { 

        var db1 = db.getSiblingDB(chunk.ns.split(".")[0]); // get the database we will be running the command against later
        var key = db.getSiblingDB("config").collections.findOne({_id:chunk.ns}).key; // will need this for the dataSize call
        // dataSize returns the info we need on the data, but using the estimate option to use counts is less intensive
        var dataSizeResult = db1.runCommand({datasize:chunk.ns, keyPattern:key, min:chunk.min, max:chunk.max, estimate:est});
        // printjson(dataSizeResult); // uncomment to see how long it takes to run and status           
        print(chunk._id+","+chunk.shard+","+dataSizeResult.size+","+dataSizeResult.numObjects); 
        totalSize += dataSizeResult.size;
        totalChunks++;
        if (dataSizeResult.size == 0) { totalEmpty++ }; //count empty chunks for summary
        }
    )
    print("***********Summary Chunk Information***********");
    print("Total Chunks: "+totalChunks);
    print("Average Chunk Size (bytes): "+(totalSize/totalChunks));
    print("Empty Chunks: "+totalEmpty);
    print("Average Chunk Size (non-empty): "+(totalSize/(totalChunks-totalEmpty)));
}

It's pretty basic at the moment, but it does the job. I have also added it on github and may expand it further there. For now though, it will do what is needed. On the test data set described at the start, the output looks like this (some data removed for brevity):

mongos> AllChunkInfo("chunkTest.foo", true);
ChunkID,Shard,ChunkSize,ObjectsInChunk
chunkTest.foo-_id_MinKey,shard0000,0,0
chunkTest.foo-_id_0.0,shard0000,599592,10707
chunkTest.foo-_id_10707.0,shard0000,1147832,20497
chunkTest.foo-_id_31204.0,shard0000,771568,13778
chunkTest.foo-_id_44982.0,shard0000,771624,13779
// omitted some data for brevity
chunkTest.foo-_id_940816.0,shard0000,1134224,20254
chunkTest.foo-_id_961070.0,shard0000,1145032,20447
chunkTest.foo-_id_981517.0,shard0000,1035104,18484
***********Summary Chunk Information***********
Total Chunks: 41
Average Chunk Size (bytes): 1365855.024390244
Empty Chunks: 1
Average Chunk Size (non-empty): 1400001.4

To explain the arguments passed to the function:

The first argument is the namespace to examine (a string), and the second (a boolean) is whether or not to use the estimate option or not. For any production environment it is recommended to use estimate:true - if it is not used, the all the data will need to be examined, and that means pulling it into memory, which will be expensive.

While the estimate:true version is not free (it uses counts and average object sizes), it is at least reasonable to run even on a large data set. The estimate version can also be a little off if object size is skewed on some shards and hence the average size is not representative (this is generally pretty rare).

MongoDB Sharded Cluster: Balancing chunks not working, stuck at “step 2”

The definition of the steps referenced in a chunk migration are a little fluid, and can depend on the version you are running. However, based on what you have provided I suspect that the primary on rs01 and the number of chunks you are trying to move is the source of the issue here.

Last time I looked at this in detail (around version 2.6), step 2 was basically a "sanity check" on the source shard primary to make sure it was ready to kick off another migration. There are two common reasons (and some not so common) for the primary to get "stuck" in step 2:

The primary is "too busy" - i.e. it is processing a lot of deletes from previous migrations and does not want to take on more migrations until some of those finish
There is an error being hit (chunk too big for example) and the migration is being aborted

Given that your two existing shards were not yet balanced when you added a third shard, though rs03 will now be the new preferred destination for migrating chunks since it has the lowest total, adding a third shard will not make this happen any more quickly. In fact it just means there are more chunks to move than before you added the third shard.

To confirm what the root cause is here, you will need to take a look at the logs on the primary of shard rs01. The shard with the most chunks will always be chosen as the source of a migration, and the primary will be the one doing all the work. The logs should tell you why the migrations are being aborted or taking a long time.

If it is a case of being too busy, you do have an option that will free it up, temporarily at least (it may get stuck again later). You can step down the primary, thereby killing any background delete jobs it is running for cleanup.

However, this will mean leaving behind "orphaned chunks" and you will need the clean up command to fix that later - you are not solving anything per se, you are just putting the work off until later. There are also only limited options to tweak this behavior, for more see here.

There are other options in terms of balancing chunks more quickly (mitosis), but they are way beyond the scope of an answer on this site I'm afraid.

If there is an error being hit, and the cleanup is not the issue, then it would probably be best to get the primary log messages and post a separate question to figure out the cause.