Mongodb – Must data be evenly distributed among Mongo shards

mongodbsharding

According to this article: https://www.mongodb.com/blog/post/on-selecting-a-shard-key-for-mongodb. Assuming I have my schema as:

{
    time_posted: ...,
    userid: ...,
    content: ...
}

The author mentions that if I frequently query the latest 10 articles from a certain user, I should set a compound key of time_posted and user_id.

But in my case, I only need to query the latest 10 articles from all users. Therefore I think I only need to set a single key time_posted. I observed that by doing so the data are mostly inserted into 1 shard, and so Read Locality is improved, which is the desired effect.

However, according to https://docs.mongodb.org/v3.0/core/sharding-balancing/, Mongo will try to balance the data evenly among all the shards.

Balancing is the process MongoDB uses to distribute data of a sharded
collection evenly across a sharded cluster. When a shard has too many
of a sharded collection’s chunks compared to other shards, MongoDB
automatically balances the chunks across the shards.

So my question are: must the data be evenly distributed among Mongo shards? and can I turn off the rebalancer since I don't need my data to be evenly distributed?

Best Answer

So even if you have 1M users, only 10 articles must be kept? I have a strong feeling that your data model has this or that flaw.

Does sharding a collection with 10 documents (give and take) make sense?

If you only need 10 documents in a collection, you don't need to shard it, since it is not going to be balanced anyway, unless the documents are exceeding around 4.8Mb in size. This size computes like this:

maxChunkSize/number of your documents is the size pre document when you chunk gets split latest, equalling 6.4Mb per document
In order to cross the migration threshold, which is only 2 for collections with less than 20 documents, we need to have your chunk split at least once to trigger the migration
Since the chunk split may be already triggered when the chunk reaches half of it's max size resulting in a split when all documents are slightly over 3.2Mb, we add the median of the difference between half of the chunks max size and and it's max size (silently assuming that a chunk is guaranteed to be split at it's max size)

But sharding this collection would not make sense the first place, as , assuming the 10 documents are a hard limit, the collection's max size can only be 160Mb as per MongoDB's BSON size limit

Do I need to have my data balanced?

Let us find out wether it is a good idea to have a bad shard key and a disabled balancer. First, if you disable the cluster balancer, this affects all sharded collections. Let us take the user collection as an example:

We have 1M users
Their user id is stored in _id, starts with 1 and each new user's _id is a simple increment
We have two shards
We disable the balancer.

Now, you shard the collection from the start. What happens internally is that there are two chunks created. In the first chunk, created on the first shard, the _id from -∞ to _id < 0 are stored. In the second chunk, created on the second shard, the _ids from 0 to +∞ are stored. Now here comes the thing: since our _id increments from 1 for each user, there is never a single user stored in the first chunk and subsequently not on the first shard. No disk space of the first chunks utilized and – with more immediate importance – no RAM, too. Since indices are (tried to be) kept in RAM along with the recently used data (called working set) amongst other things to speed operations up, sooner or later we will have the situation that the RAM on the first shard is rather empty, while the second shard will start to evict data set items or even indices out of RAM. Bad idea, huh? Now things get worse: Since we have disabled the balancer, the cluster can do nothing to mitigate the situation.

Now let's assume we were slightly smarter and have pre-split our chunks so that the _id for our users collection are distributed in a way where the _ids ranging from -∞ to 500,000 are stored on the first shard and the rest on the second. It is obvious that this is only a temporary solution, since when we exceed 1M users, the whole problem starts again. And without the balancer running, the cluster still can not mitigate this situation.

Taking this a step further: We have found out that we can use a hash sum of our _id as our shard key. Jay! Problem solved! Except it isn't. In theory, the hash algorithm should cause our users to be evenly distributed among our shards. But there is a little thing called variance.

It can be easily demonstrated like this (heavily simplified for the sake of shortness): When tossing a coin 20 times, in theory you should get an equal number of heads and tails, since the probability of either of them is 50% right? Try it. Now. The odds are very low that you will have an equal amount of heads and tails.

How does this translate to our problem? Well, chances are high that either the first or the second shard get's slightly more documents than the other over time. And this sooner or later will add up to a point where it becomes a problem – RAM and disk space is underutilized on one shard and over utilized on the other. Again. And again, the balancer can not help us to mitigate the problem.

So this should have made it crytsal clear that you should have your data balanced and that you should never have the balancer disabled by default (there are some administrative tasks during which you should or must have the balancer disabled).

Conclusion

For some 10 documents, no matter how big they are, you don't even need to shard the collection.

Disabling the balancer might cause severe problems with your cluster. Unless you absolutely have to or you are absolutely positively sure that you can live with the consequences, do not disable it.

Please note that I left out some more complicated topics, like hard drive IO bandwidth bottlenecks, network bandwidth distribution and alike for the sake of readability and – as funny as this might sound – shortness.

Related Solutions

Mongodb – Are shards distributed randomly across machines in MongoDB or are they in order

In general, no they are not distributed in order. However, there is a way that this (or at least approximately this) distribution can occur during normal operations. To explain:

If you have all of your data on a single shard (a-z) and then add a second shard, the balancer will redistribute the data to the new shard by migrating chunks.

The way the balancer picks a chunk to migrate in general is pretty simple, it will pick the lowest available range on the shard with the most chunks (the "a" chunk for example), and move it to the shard with the lowest number of chunks. Rinse and repeat until the chunks are balanced.

Hence, if you have all your data on one shard and add a second shard, the low range chunks (a-m) will all be moved from the original shard to the new shard, and you can end up with this almost-in-order distribution.

In your example, you just end up with an oddly uniform distribution of data, no real harm done. However, it might be problematic from a performance perspective for range based queries across a large portion of the data. Think about a query that would walk the index in order - it would only ever be hitting one shard (new shard, then old shard) at a time rather than using the resources of both with better distribution of data.

In other scenarios this behavior can lead to a particularly poor data distribution. Let's say you had a time based shard key (bad for various reasons, but still quite common) and you deleted old data on a regular basis. Combine that with the scenario above and the new shard would eventually have no data on it (the data in the low range, old, chunks would all be deleted). Remember that an empty chunk (no data) counts just as much as a 60MB chunk in terms of balancing (the balancer is purely looking at the number of chunks on each shard, not the amount of data in them).

Also, none of this would happen if you had 2 shards to start with and then inserted your data, because the splits and migrations would be more random and more distributed in general. MongoDB will also attempt to move the maximum chunk as well to help this distribution happen.

To summarize - if you have spotted this type of distribution problem, you can take steps to fix it by moving chunks around yourself. Take a look at the moveChunk command (it's generally a good idea, but not required, to have the balancer off when you do this or you can have trouble getting the required lock).

The easiest way to view all this is to run the sharding status command and pass and argument of true, i.e. sh.status(true). This will print details of all sharded collections, where the chunks live etc. It pulls its information from the config database and uses a simple Map Reduce to aggregate the information.

Armed with that information you can also inspect the changelog collection to determine what has been migrated and when it was done (this will also be recorded in your primary mongod logs but can be tough to parse out on busy systems).

Finally, I realize this is somewhat beyond the scope of this answer, but I have written a couple of Javascript functions (that you can use from the MongoDB shell) that will let you determine the object and data distribution on your shards for a given collection, something not easily available elsewhere. Hopefully you will find them useful :)

Please Note: these functions can be intensive to run, so use with caution in a production environment.

This first function prints out the chunk information in CSV format for a given namespace. It uses the estimate option in the datasize command to use counts (rather than a scan of all the objects) to figure out the size of each chunk. Removing the estimate option would give a more accurate result (particularly if your document size varies) but if the data set is not in RAM it can be very resource intensive.

AllChunkInfo = function(ns){
    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:true});
        // 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)));
}

You can call this on any sharded namespace from a mongos as follows:

mongos> AllChunkInfo("test.users");
ChunkID,Shard,ChunkSize,ObjectsInChunk
test.users-_id_MinKey,shard0000,0,0
test.users-_id_10000.0,shard0000,525420,26271
test.users-_id_36271.0,shard0001,1120640,56032
test.users-_id_92303.0,shard0001,153940,7697
***********Summary Chunk Information***********
Total Chunks: 4
Average Chunk Size (bytes): 450000
Empty Chunks: 1
Average Chunk Size (non-empty): 600000

The next lets you find the information for a particular chunk, again using an estimate. This is somewhat safer to use without the estimate, but should still be used with caution:

ChunkInfo = function(ns, id){
    var configDB = db.getSiblingDB("config");
    var db1 = db.getSiblingDB(ns.split(".")[0]);
    var key = configDB.collections.findOne({_id:ns}).key;
    var chunk = configDB.chunks.find({"_id" : id, }).limit(1).next();
    var dataSizeResult = db1.runCommand({datasize:chunk.ns, keyPattern:key, min:chunk.min, max:chunk.max, estimate:true});
    print("***********Chunk Information***********");
    printjson(chunk);
    print("Chunk Size: "+dataSizeResult.size)
    print("Objects in chunk: "+dataSizeResult.numObjects)
}

Sample Output:

mongos> ChunkInfo("test.users", "test.users-_id_10000.0")
***********Chunk Information***********
{
        "_id" : "test.users-_id_10000.0",
        "lastmod" : Timestamp(3000, 0),
        "lastmodEpoch" : ObjectId("518373f0a962406e4467b054"),
        "ns" : "test.users",
        "min" : {
                "_id" : 10000
        },
        "max" : {
                "_id" : 36271
        },
        "shard" : "shard0000"
}
Chunk Size: 525420
Objects in chunk: 26271

Mongodb – Sharding Mongodb while limiting the number of documents to be migrated

For your 7 collections on the primary shard:

Enter as administrator on the primary replica (if you shard your replica) and create your collection there by inserting 1 document or creating an index there. When you create a collection via the mongos - shard client - then the collection is started on a random shard; if you create it on the shard itself first, then you know it's on the one you want. shard1$ mongo --port 27018 localhost/mydb --eval 'db.mycol.insert({firstdocument:'hello'})'
For the 25 collections:

Standard sharding indeed as you mention. Be patient with the distribution. The first shards will start distributing after a certain amount of data, can be 100k documents if the documents are small.
For the big collection:

Have a look at shard-tags. So, you have to tag your collection on shard1 with let's say : {status:'in_use'}; shard2 is tagged {status:'freeBeer'}. Then depending on the status in the document, the shard is chosen.
For the disk issue, I think you gonna have to write a script. Get a warning when 60% of a disk is used, and another warning at 80% to take action. Then you'll need to upgrade your disk or redistribute the chunks. Certainly the 2TB collection could give you space via the tag, to send more to shard2. Or, add a shard3 ... what is actually the purpose of sharding, to grow horizontally.
Choose shardKeys well, practice configuration and setup on a small server, practice moving chunks, adding tags, ....
Have a look at the AllChunkInfo script of Adam Comerford for you scripts and testing : https://github.com/comerford/mongodb-scripts