This topic outlines MongoDB sharded clusters, which are used to store massive amounts of data.
When to use a sharded cluster?
Use a sharded cluster to solve the following problems:
Storage capacity is limited by a single host, which creates a disk space bottleneck.
Read and write performance is limited by a single host. This limitation may be caused by resource bottlenecks in the CPU, memory, or network interface controller (NIC), which prevent you from scaling performance.
How to determine the number of shards and mongos?
You can determine the number of shards and mongos nodes as follows:
The sharded cluster is used only to store a large amount of data and has a low access volume. For example, if a single shard can store M data and the total required storage is N, you can calculate the required number of shards and mongos nodes using these formulas:
numberOfShards = N/M/0.75 (Assumes a storage capacity watermark of 75%)
numberOfMongos = 2+ (For low access requirements, deploy at least two mongos nodes for high availability)
The sharded cluster is used to handle high-concurrency writes or reads, but the total data volume is small. The shards and mongos nodes must meet the read and write performance requirements. For example, if the maximum queries per second (QPS) of a single shard is M, the maximum QPS of a single mongos is Ms, and the total required QPS is Q, you can calculate the required number of shards and mongos nodes using these formulas:
numberOfShards = Q/M/0.75 (Assumes a load watermark of 75%)
numberOfMongos = Q/Ms/0.75
NoteIf the sharded cluster must solve both problems, you can estimate the number of nodes based on the higher requirement.
These formulas provide an estimate for an ideal scenario where data and requests are evenly distributed. In practice, the distribution may be uneven. To balance the system load as evenly as possible, you must select a suitable shard key.
The service capabilities of mongos and mongod must be measured based on your specific access patterns.
Shard key selection
Sharding strategies supported by MongoDB sharded clusters
Range sharding supports range queries based on the shard key.
Hash sharding evenly distributes writes across shards.
Tag aware sharding lets you customize chunk distribution rules.
NoteHow it works
Use
sh.addShardTag()to set Tag A on a shard.Use
sh.addTagRange()to set Tag A on a chunk range of a collection. MongoDB then ensures that the chunk range with Tag A, or a superset of that range, is distributed on the shard with Tag A.
Scenarios
You can set data center tags on shards deployed in different data centers. This distributes data from different chunk ranges to the specified data centers.
You can set service level tags on shards with different service capabilities. This distributes more chunks to shards with higher capabilities.
Notes
ApsaraDB for MongoDB cannot directly distribute chunks to shard nodes that are labeled with the specified tags. Chunks are gradually distributed as chunk splitting and migration are frequently triggered by insert and update operations. Make sure that the balancer is enabled during the distribution process. After tags are added to chunk ranges, written data may not be distributed to the shard node that has the same tags as the chunk ranges.
Problems that range and hash sharding cannot solve
The value range of the shard key is too small. For example, if you use a data center as the shard key, sharding is not effective because there are usually few data centers.
A specific shard key value is present in too many documents. This can lead to a single, oversized chunk, known as a jumbo chunk, that affects chunk migration and load balancing.
Queries and updates based on a non-shard key field become scatter-gather queries, which are inefficient.
Characteristics of a good shard key
Sufficient cardinality
Evenly distributed writes
Avoid scatter-gather queries (targeted reads).
Example:
An IoT application uses a sharded cluster instance to store the logs of millions of devices. Each device sends logs to the sharded cluster instance once every 10 seconds. The logs contain information such as device IDs and timestamps. The logs that are generated for a specific device over a specific time range are frequently queried.
Query request: Queries logs for a specific device within a specified time range.
(Recommended) Method 1: Use a compound shard key of device ID and timestamp with range sharding.
Writes are evenly distributed across multiple shards.
Data for the same device ID is further distributed across multiple chunks based on the timestamp.
Queries for a time range by device ID can be completed directly using the (deviceId, timestamp) compound index.
Method 2: Use the timestamp as the shard key with range sharding.
New writes have consecutive timestamps and are all sent to the same shard, causing uneven write distribution.
Queries by device ID are scattered to all shards, which is inefficient.
Method 3: Use the timestamp as the shard key with hash sharding.
Writes are evenly distributed across multiple shards.
Queries by device ID are scattered to all shards, which is inefficient.
Method 4: Use the device ID as the shard key with hash sharding.
NoteIf the device ID has no obvious pattern, you can use range sharding.
Writes are evenly distributed across multiple shards.
Data for the same device ID cannot be further subdivided and can only be distributed to a single chunk. This causes jumbo chunks. Queries by device ID are sent to a single shard. After the request is routed to that shard, a range query based on the timestamp requires a full table scan and a sort.
Jumbo chunk and size per chunk
The default size per chunk in a sharded cluster instance is 64 MB. If the size of a chunk exceeds 64 MB and the chunk cannot be split, the chunk is labeled as a jumbo chunk. For example, if all documents have the same shard key value, these documents are stored in the same chunk, which cannot be split. The balancer cannot migrate jumbo chunks, which may cause load imbalance. We recommend that you prevent jumbo chunks.
If a jumbo chunk appears and you do not have high requirements for load balancing, it will not affect data reads and writes. You can use one of the following methods to handle it:
You can split the jumbo chunk. After the split is successful, mongos automatically clears the jumbo flag.
For a chunk that cannot be split, if it is no longer a jumbo chunk, you can try to manually clear the jumbo flag.
NoteBefore performing a purge, you must back up the `config` database to prevent corruption from operational errors.
You can increase the chunk size. When the chunk size is no longer smaller than the configured chunk size, the jumbo flag is eventually cleared. However, jumbo chunks may still appear as data is written. The fundamental solution is to plan your shard key properly.
Scenarios for adjusting the chunk size (value range: 1 to 1024 MB):
If the I/O load is too high during migration, you can try setting a smaller chunk size.
During testing, you can set a smaller chunk size to easily verify the results.
If the initial chunk size setting is unreasonable and causes many jumbo chunks that affect load balancing, you can try increasing the chunk size.
When converting an unsharded collection to a sharded collection, if the collection is very large, you may need to increase the chunk size for the conversion to succeed. You may encounter this with terabyte-level data volumes. For more information, see Sharding Existing Collection Data Size.
About load balancing
Automatic load balancing in a sharded cluster instance is implemented by a background thread that runs on the mongos nodes of the instance. Only one migration task can be run for each collection at a specific point in time. Load balancing is triggered based on the number of chunks per collection on each shard node. If the difference between the number of chunks of a collection on a shard node and the total number of chunks of the collection reaches the specified threshold, ApsaraDB for MongoDB starts to migrate the chunks of the collection from the shard node to other shard nodes. You must specify a threshold based on the total number of chunks.
Load balancing is enabled by default. To prevent chunk migration from affecting online services, you can set a migration window. For example, you can allow migration only between 02:00 and 06:00.
use config
db.settings.update(
{ _id: "balancer" },
{ $set: { activeWindow : { start : "02:00", stop : "06:00" } } },
{ upsert: true }
)
When you back up a sharded cluster, either through mongos or by backing up the Configserver and all shards separately, you must run the following command to stop the balancer. This prevents status inconsistencies in the backup data.
sh.stopBalancer() Archiving setting in the moveChunk command
If a sharded cluster instance runs MongoDB 3.0 or earlier, the disk space usage in the data catalog may continue to increase even after data writes are stopped.
This problem is caused by the sharding.archiveMovedChunks configuration item, which defaults to true in MongoDB 3.0 and earlier. This setting means that during a moveChunk operation, the source shard archives the migrated chunk data for recovery purposes. As a result, when a chunk is migrated, the disk space on the source node is not released, while the destination node uses new disk space.
In MongoDB 3.2, the default value for this configuration item is false. By default, data from a moveChunk operation is not archived on the source shard.