Ascending Shard Keys

Ascending shard keys are generally something like a "date" field or ObjectId—anything that steadily increases over time. An autoincrementing primary key is another example of an ascending field, albeit one that doesn’t show up in MongoDB much (unless you’re importing from another database).

Suppose that we shard on an ascending field, like "_id" on a collection using ObjectIds. If we shard on "_id", then the data will be split into chunks of "_id" ranges, as in Figure 16-1. These chunks will be distributed across our sharded cluster of, let’s say, three shards, as shown in Figure 16-2.

Figure 16-1. The collection is split into ranges of ObjectIds; each range is a chunk

Suppose we create a new document. Which chunk will it be in? The answer is the chunk with the range ObjectId("5112fae0b4a4b396ff9d0ee5") through $maxKey. This is called the max chunk, as it is the chunk containing $maxKey.

If we insert another document, it will also be in the max chunk. In fact, every subsequent insert will be into the max chunk! Every insert’s "_id" field will be closer to infinity than the previous one (because ObjectIds are always ascending), so they will all go into the max chunk.

Figure 16-2. Chunks are distributed across shards in a random order

This has a couple of interesting (and often undesirable) properties. First, all of your writes will be routed to one shard (shard0002, in this case). This chunk will be the only one growing and splitting, as it is the only one that receives inserts. As you insert data, new chunks will “fall off” of this chunk, as shown in Figure 16-3.

Figure 16-3. The max chunk continues growing and being split into multiple chunks

This pattern often makes it more difficult for MongoDB to keep chunks evenly balanced because all the chunks are being created by one shard. Therefore, MongoDB must constantly move chunks to other shards instead of correcting the small imbalances that might occur in more evenly distributed systems.

Randomly Distributed Shard Keys

At the other end of the spectrum are randomly distributed shard keys. Randomly distributed keys could be usernames, email addresses, UUIDs, MD5 hashes, or any other key that has no identifiable pattern in your dataset.

Suppose the shard key is a random number between 0 and 1. We’ll end up with a random distribution of chunks on the various shards, as shown in Figure 16-4.

Figure 16-4. As in the previous section, chunks are distributed randomly around the cluster

As more data is inserted, the data’s random nature means that inserts should hit every chunk fairly evenly. You can prove this to yourself by inserting 10,000 documents and seeing where they end up:

> var servers = {}
> var findShard = function (id) {
...     var explain = db.random.find({_id:id}).explain();
...     for (var i in explain.shards) {
...         var server = explain.shards[i][0];
...         if (server.n == 1) {
...             if (server.server in servers) {
...                 servers[server.server]++;
...             } else {
...                 servers[server.server] = 1;
...             }
...         }
...     }
... }
> for (var i = 0; i < 10000; i++) {
...     var id = ObjectId();
...     db.random.insert({"_id" : id, "x" : Math.random()});
...     findShard(id);
... }
> servers
{
    "spock:30001" : 2942,
    "spock:30002" : 4332,
    "spock:30000" : 2726
}

As writes are randomly distributed, the shards should grow at roughly the same rate, limiting the number of migrates that need to occur.

The only downside to randomly distributed shard keys is that MongoDB isn’t efficient at randomly accessing data beyond the size of RAM. However, if you have the capacity or don’t mind the performance hit, random keys nicely distribute load across your cluster.

Location-Based Shard Keys

Location-based shard keys may be things like a user’s IP, latitude and longitude, or address. They’re not necessarily related to a physical location field: the “location” might be a more abstract way that data should be grouped together. In any case, a location-based key is a key where documents with some similarity fall into a range based on this field. This can be handy for both putting data close to its users and keeping related data together on disk. It may also be a legal requirement to remain compliant with GDPR or other similar data privacy legislation. MongoDB uses Zoned Sharding to manage this.

For example, suppose we have a collection of documents that are sharded on IP address. Documents will be organized into chunks based on their IPs and randomly spread across the cluster, as shown in Figure 16-5.

Figure 16-5. A sample distribution of chunks in the IP address collection

If we wanted certain chunk ranges to be attached to certain shards, we could zone these shards and then assign chunk ranges to each zone. In this example, suppose that we wanted to keep certain IP blocks on certain shards: say, 56.*.*.* (the United States Postal Service’s IP block) on shard0000 and 17.*.*.* (Apple’s IP block) on either shard0000 or shard0002. We do not care where the other IPs live. We could request that the balancer do this by setting up zones:

> sh.addShardToZone("shard0000", "USPS")
> sh.addShardToZone("shard0000", "Apple")
> sh.addShardToZone("shard0002", "Apple")

Next, we create the rules:

> sh.updateZoneKeyRange("test.ips", {"ip" : "056.000.000.000"}, 
... {"ip" : "057.000.000.000"}, "USPS")

This attaches all IPs greater than or equal to 56.0.0.0 and less than 57.0.0.0 to the shard zoned as "USPS". Next, we add a rule for Apple:

> sh.updateZoneKeyRange("test.ips", {"ip" : "017.000.000.000"}, 
... {"ip" : "018.000.000.000"}, "Apple")

When the balancer moves chunks, it will attempt to move chunks with those ranges to those shards. Note that this process is not immediate. Chunks that were not covered by a zone key range will be moved around normally. The balancer will continue attempting to distribute chunks evenly among shards.

Shard Key Limitations

Shard keys cannot be arrays. sh.shardCollection() will fail if any key has an array value, and inserting an array into that field is not allowed.

Once inserted, a document’s shard key value may be modified unless the shard key field is an immutable _id field. In older versions of MongoDB prior to 4.2, it was not possible to modify a document’s shard key value.

Most special types of indexes cannot be used for shard keys. In particular, you cannot shard on a geospatial index. Using a hashed index for a shard key is allowed, as covered previously.

Shard Key Cardinality

Whether your shard key jumps around or increases steadily, it is important to choose a key with values that will vary. As with indexes, sharding performs better on high-cardinality fields. If, for example, you had a "logLevel" key that had only values "DEBUG", "WARN", or "ERROR", MongoDB wouldn’t be able to break up your data into more than three chunks (because there would be only three different values for the shard key). If you have a key with little variation and want to use it as a shard key anyway, you can do so by creating a compound shard key on that key and a key that varies more, like "logLevel" and "timestamp". It is important that the combination of keys has high cardinality.

Using a Cluster for Multiple Databases and Collections

MongoDB evenly distributes collections across every shard in your cluster, which works well if you’re storing homogeneous data. However, if you have a log collection that is “lower value” than your other data, you might not want it taking up space on your more expensive servers. Or, if you have one powerful shard, you might want to use it for only a real-time collection and not allow other collections to use it. You can create separate clusters, but you can also give MongoDB specific directions about where you want it to put certain data.

To set this up, use the sh.addShardToZone() helper in the shell:

> sh.addShardToZone("shard0000", "high")
> // shard0001 - no zone
> // shard0002 - no zone
> // shard0003 - no zone
> sh.addShardToZone("shard0004", "low")
> sh.addShardToZone("shard0005", "low")

Then you can assign different collections to different shards. For instance, for your super-important real-time collection:

> sh.updateZoneKeyRange("super.important", {"<shardKey>" : MinKey}, 
... {"<shardKey>" : MaxKey}, "high")

This says, “for negative infinity to infinity for this collection, store it on shards tagged "high".” This means that no data from the super.important collection will be stored on any other server. Note that this does not affect how other collections are distributed: they will still be evenly distributed between this shard and the others.

You can perform a similar operation to keep the log collection on a low-quality server:

> sh.updateZoneKeyRange("some.logs", {"<shardKey>" : MinKey}, 
... {"<shardKey>" : MaxKey}, "low")

The log collection will now be split evenly between shard0004 and shard0005.

Assigning a zone key range to a collection does not affect it instantly. It is an instruction to the balancer stating that, when it runs, these are the viable targets to move the collection to. Thus, if the entire log collection is on shard0002 or evenly distributed among the shards, it will take a little while for all of the chunks to be migrated to shard0004 and shard0005.

As another example, perhaps you have a collection that you don’t want on the shard zoned "high", but you don’t care which other shard it goes on. You can zone all of the non-high-performance shards to create a new grouping. Shards can have as many zones as you need:

> sh.addShardToZone("shard0001", "whatever")
> sh.addShardToZone("shard0002", "whatever")
> sh.addShardToZone("shard0003", "whatever")
> sh.addShardToZone("shard0004", "whatever")
> sh.addShardToZone("shard0005", "whatever")

Now you can specify that you want this collection (call it normal.coll) distributed across these five shards:

> sh.updateZoneKeyRange("normal.coll", {"<shardKey>" : MinKey}, 
... {"<shardKey>" : MaxKey}, "whatever")

If you make a mistake or change your mind, you can remove a shard from a zone with sh.removeShardFromZone():

> sh.removeShardFromZone("shard0005", "whatever")

If you remove all shards from zones described by a zone key range (e.g., if you remove shard0000 from the zone "high"), the balancer won’t distribute the data anywhere because there aren’t any valid locations listed. All the data will still be readable and writable; it just won’t be able to migrate until you modify your tags or tag ranges.

To remove a key range from a zone, use sh.removeRangeFromZone(). The following is an example. The range specified must be an exact match to a range previously defined for the namespace some.logs and a given zone:

> sh.removeRangeFromZone("some.logs", {"<shardKey>" : MinKey}, 
... {"<shardKey>" : MaxKey})

Manual Sharding

Sometimes, for complex requirements or special situations, you may prefer to have complete control over which data is distributed where. You can turn off the balancer if you don’t want data to be automatically distributed and use the moveChunk command to manually distribute data.

To turn off the balancer, connect to a mongos (any mongos is fine) using the mongo shell and disable the balancer using the shell helper sh.stopBalancer():

> sh.stopBalancer()

If there is currently a migrate in progress, this setting will not take effect until the migrate has completed. However, once any in-flight migrations have finished, the balancer will stop moving data around. To verify no migrations are in progress after disabling, issue the following in the mongo shell:

> use config
> while(sh.isBalancerRunning()) {
...  print("waiting...");
...  sleep(1000);
... }

Once the balancer is off, you can move data around manually (if necessary). First, find out which chunks are where by looking at config.chunks:

> db.chunks.find()

Now, use the moveChunk command to migrate chunks to other shards. Specify the lower bound of the chunk to be migrated and give the name of the shard that you want to move the chunk to:

> sh.moveChunk(
... "test.manual.stuff", 
... {user_id: NumberLong("-1844674407370955160")}, 
... "test-rs1")

However, unless you are in an exceptional situation, you should use MongoDB’s automatic sharding instead of doing it manually. If you end up with a hotspot on a shard that you weren’t expecting, you might end up with most of your data on that shard.

In particular, do not combine setting up unusual distributions manually with running the balancer. If the balancer detects an uneven number of chunks it will simply reshuffle all of your work to get the collection evenly balanced again. If you want uneven distribution of chunks, use the zone sharding technique discussed in “Using a Cluster for Multiple Databases and Collections”.