Learn MongoDB sharding concepts, implementation, configuration, and best practices. Understand how to scale horizontally with MongoDB sharded clusters.
Learn MongoDB sharding concepts, implementation, configuration, and best practices. Understand how to scale horizontally with MongoDB sharded clusters.
is MongoDB's approach to horizontal scaling, which distributes data across multiple machines
to support deployments with very large datasets and high throughput operations. As data volume and throughput requirements
increase, a single server or replica set can become a bottleneck. Sharding solves this by horizontally scaling the database
across multiple servers, allowing MongoDB to handle datasets larger than a single server's capacity.
In this lesson, we'll explore the core concepts of MongoDB sharding, how to set up and configure a sharded cluster,
sharding strategies, and best practices for production deployments.
Unlike replication, which provides high availability and redundancy, sharding focuses on horizontal scalability
by distributing data across multiple machines. When implemented correctly, sharding allows MongoDB to handle virtually
unlimited data growth and throughput requirements.
A MongoDB sharded cluster consists of three main components:
Shard Servers
: Each shard is a replica set that holds a subset of the sharded data.
In production environments, each shard should be a replica set to provide redundancy and high availability.
Config Servers
: These store the cluster's metadata, including the mapping of data chunks
to shards. Config servers are deployed as a replica set to ensure metadata availability.
Router Servers (mongos)
: These instances route queries and write operations to the appropriate
shard(s). Applications connect to the mongos routers, which act as a query router, providing a unified interface
to the sharded cluster.
When a client sends a request to a mongos router:
The mongos determines which shard(s) contains the relevant data using metadata from the config servers.
It routes the operation to the appropriate shard(s).
It collects and merges the results if necessary.
It returns the final results to the client.
Sharding Architecture Diagram
Client Applications
[mongos] → [mongos] → [mongos] Query Routers
↓ ↓ ↓
[Config Servers Replica Set] Metadata & Configuration
↓ ↓ ↓
[Shard 1] [Shard 2] [Shard 3] Data Shards (Each a Replica Set)
Learn about Sharding Concepts.
is a field or combination of fields that MongoDB uses to distribute documents among
shards. The choice of shard key is critical as it directly impacts the cluster's performance, scalability, and
balancing characteristics.
Key considerations when choosing a shard key include:
Cardinality
: A higher number of possible values allows for better data distribution.
: Values should appear with similar frequency to prevent hotspots.
Rate of change
: Shard key values should not increase or decrease monotonically (causing all new writes to go to one shard).
Query patterns
: The shard key should be frequently included in queries to enable targeted operations.
MongoDB divides sharded data into
. Each chunk contains a subset of sharded data and has an
inclusive lower bound and an exclusive upper bound based on the shard key. Initially, a sharded collection has a single
chunk covering the entire range of possible shard key values.
As data grows, MongoDB splits chunks that exceed the configured size (default 64 MB) to maintain evenly sized chunks.
The balancer process then migrates chunks across shards to ensure an even distribution.
is a background process that runs in the MongoDB cluster to ensure an even distribution
of chunks across all shards. When it detects an imbalance, it initiates chunk migrations to move chunks from shards with
more chunks to those with fewer.
During migration:
The balancer initiates a migration from source to target shard.
The source shard transfers documents in the chunk to the target shard.
The source continues to forward operations for the migrating chunk to the target.
Once completed, the config servers update the metadata to reflect the new chunk location.
MongoDB offers three sharding strategies that determine how data is distributed across shards:
Ranged sharding
divides data into contiguous ranges determined by the shard key values.
This strategy works well for queries that need to retrieve documents within a specific range of shard key values.
Advantages
: Supports efficient range queries on the shard key.
Disadvantages
: Can lead to uneven data distribution if the shard key values aren't evenly distributed.
1
2// Enabling ranged sharding on a collection
3sh.shardCollection(
4 "mydb.users",
5 { lastName: 1, firstName: 1 } // Shard key
6)
7 Hashed sharding
uses a hash of the shard key's value to distribute data more randomly.
This provides a more even data distribution but makes range-based queries less efficient.
Advantages
: More even data distribution, especially with monotonically changing keys.
Disadvantages
: Range-based queries on the shard key become scatter-gather operations.
1
2// Create a hashed index
3db.users.createIndex({ userId: "hashed" })
4
5// Enable hashed sharding
6sh.shardCollection(
7 "mydb.users",
8 { userId: "hashed" } // Hashed shard key
9)
10 Zoned sharding
allows you to associate specific ranges of shard key values with specific shards,
called zones. This gives you more control over data placement, which is useful for geographic distribution or
hardware tiering.
Advantages
: Precise control over data locality, useful for compliance or performance optimization.
Disadvantages
: More complex to configure and maintain.
1
2// Define zones
3sh.addShardToZone("shard0001", "US")
4sh.addShardToZone("shard0002", "EU")
5sh.addShardToZone("shard0003", "ASIA")
6
7// Associate ranges of shard key values with zones
8sh.updateZoneKeyRange(
9 "mydb.users",
10 { "region": "US", "userId": MinKey },
11 { "region": "US", "userId": MaxKey },
12 "US"
13)
14
15sh.updateZoneKeyRange(
16 "mydb.users",
17 { "region": "EU", "userId": MinKey },
18 { "region": "EU", "userId": MaxKey },
19 "EU"
20)
21 Setting up a sharded cluster involves several steps:
Learn about 1. Set Up Config Server Replica Set.
1
2# Start each config server
3mongod --configsvr --replSet configRS --port 27019 --dbpath /data/configdb
4
5# On one config server, initiate the replica set
6rs.initiate({
7 _id: "configRS",
8 configsvr: true,
9 members: [
10 { _id: 0, host: "config1:27019" },
11 { _id: 1, host: "config2:27019" },
12 { _id: 2, host: "config3:27019" }
13 ]
14})
15 Learn about 2. Set Up Shard Replica Sets.
1
2# Start each shard server
3mongod --shardsvr --replSet shard1RS --port 27018 --dbpath /data/shard1
4
5# On one server of each shard, initiate the replica set
6rs.initiate({
7 _id: "shard1RS",
8 members: [
9 { _id: 0, host: "shard1-a:27018" },
10 { _id: 1, host: "shard1-b:27018" },
11 { _id: 2, host: "shard1-c:27018" }
12 ]
13})
14
15# Repeat for each shard (shard2RS, shard3RS, etc.)
16 Learn about 3. Start Mongos Routers.
1
2# Start mongos router instances
3mongos --configdb configRS/config1:27019,config2:27019,config3:27019 --port 27017
4
5# Start additional mongos routers for redundancy
6 Learn about 4. Add Shards to the Cluster.
1
2// Connect to mongos
3mongo --host mongos1:27017
4
5// Add each shard to the cluster
6sh.addShard("shard1RS/shard1-a:27018,shard1-b:27018,shard1-c:27018")
7sh.addShard("shard2RS/shard2-a:27018,shard2-b:27018,shard2-c:27018")
8sh.addShard("shard3RS/shard3-a:27018,shard3-b:27018,shard3-c:27018")
9
10// Verify the shards
11sh.status()
12 Learn about 5. Enable Sharding for a Database and Collection.
1
2// Enable sharding for a database
3sh.enableSharding("mydb")
4
5// Create an index for the shard key
6db.users.createIndex({ userId: 1 })
7
8// Shard the collection
9sh.shardCollection("mydb.users", { userId: 1 })
10
11// Check sharding status
12sh.status()
13 The mongos router determines how to execute queries based on the shard key and query predicates:
When a query includes the shard key or a prefix of the shard key, mongos can route the operation to specific shards,
resulting in more efficient execution.
1
2// Targeted query - only goes to the shard containing documents with userId: 123
3db.users.find({ userId: 123 })
4
5// Targeted query with a compound shard key
6db.users.find({ region: "US", userId: 123 })
7
8// Targeted range query on a prefix of the shard key
9db.users.find({ region: "US" }) // If shard key is { region: 1, userId: 1 }
10 Queries that don't include the shard key must be sent to all shards. The mongos sends the query to every shard,
waits for responses, and then combines the results.
: Scatter-gather operations can be significantly slower as data volume grows, especially when combined
with sorts, limits, or skips that can't be pushed down to the individual shards.
1
2// Scatter-gather query - must be sent to all shards
3db.users.find({ age: { $gt: 30 } })
4
5// Scatter-gather query with sort and limit
6db.users.find({ age: { $gt: 30 } }).sort({ lastName: 1 }).limit(10)
7 Learn about Chunk Migration & Balancing.
The balancer process continually monitors the chunk distribution across shards. When it detects an imbalance, it
initiates migrations to evenly distribute chunks.
You can check the balancer status and configure it:
1
2// Check balancer status
3sh.getBalancerState()
4
5// Check if balancer is currently running
6sh.isBalancerRunning()
7
8// Disable the balancer (for maintenance)
9sh.stopBalancer()
10
11// Enable the balancer
12sh.startBalancer()
13
14// Configure balancing window (only balance during off-peak hours)
15db.settings.updateOne(
16 { _id: "balancer" },
17 { $set: { activeWindow: { start: "23:00", stop: "6:00" } } }
18)
19 You can monitor chunk migrations to understand the balancing activity in your cluster:
1
2// Check recent migrations
3db.getSiblingDB("config").changelog.find({ what: "moveChunk.from" })
4 .sort({ time: -1 })
5 .limit(10)
6
7// Check current migrations
8db.getSiblingDB("admin").currentOp(
9 { $or: [
10 { "op": "moveChunk" },
11 { "desc": { $regex: /^moving/ } }
12 ]
13 }
14)
15 Learn about Sharding Best Practices.
High cardinality
: Choose a shard key with many unique values to ensure even distribution.
Low frequency
: Values should appear with similar frequency to prevent hotspots.
Non-monotonic
: Avoid shard keys that increase or decrease continuously (like timestamps or auto-incrementing IDs) without hashing.
Query-aligned
: The shard key should match your common query patterns to allow targeted operations.
Start with enough shards
: Begin with at least 3-4 shards to minimize the need for extensive rebalancing as your cluster grows.
Use jumbo chunks judiciously
: Monitor for jumbo chunks (chunks too large to migrate) and split them manually if necessary.
Monitor balancing operations
: Schedule balancing during off-peak hours to minimize impact on production traffic.
Scale mongos routers
: Deploy multiple mongos routers and load balance application connections across them.
Understand query patterns
: Analyze and optimize queries that might lead to scatter-gather operations.
Changing shard keys
: Once a collection is sharded, you cannot change its shard key without dropping and recreating the collection.
Underprovisioning config servers
: Always use a 3-member replica set for config servers in production.
Neglecting shard key indexes
: Ensure your shard key fields are indexed to support efficient queries.
Forgetting about shard key in queries
: Design application queries to include the shard key whenever possible.
Effective monitoring is crucial for maintaining healthy sharded clusters:
Chunk distribution
: Monitor the number of chunks per shard to ensure balanced distribution.
Migration activity
: Track frequency and duration of chunk migrations.
Jumbo chunks
: Identify chunks that cannot be split or migrated.
Query patterns
: Monitor for scatter-gather operations that might indicate suboptimal shard key selection.
Connection counts
: Check connection distribution across mongos routers.
Learn about Useful Monitoring Commands.
1
2// Get detailed sharding status
3sh.status({ verbose: true })
4
5// Check chunk distribution
6db.getSiblingDB("config").chunks.aggregate([
7 { $group: { _id: "$shard", count: { $sum: 1 } } },
8 { $project: { shard: "$_id", count: 1, _id: 0 } },
9 { $sort: { count: -1 } }
10])
11
12// Monitor ongoing migrations
13db.getSiblingDB("admin").currentOp(
14 { $or: [
15 { "op": "moveChunk" },
16 { "desc": { $regex: /^moving/ } }
17 ]
18 }
19)
20
21// Check for jumbo chunks
22db.getSiblingDB("config").chunks.find({ "jumbo": true })
23
24// Monitor mongos routers
25db.adminCommand({ "serverStatus": 1 }).connections
26 Learn about Advanced Sharding Topics.
MongoDB supports multi-document transactions in sharded clusters, but there are some considerations:
Transactions that span multiple shards have higher latency and are more sensitive to network issues.
For best performance, design transactions to operate on documents that are on the same shard.
Use the shard key in transaction operations to enable targeted routing.
Once a collection is sharded, you cannot update the values of the shard key fields in any document.
In MongoDB 4.2+, you can update a document's shard key value by deleting and reinserting the document,
but this can cause significant overhead.
As of MongoDB 5.0, you can reshard collections (change the shard key) without downtime.
This is particularly useful when your initial shard key choice proves suboptimal as your data patterns evolve.
1
2// Starting a resharding operation (MongoDB 5.0+)
3db.adminCommand({
4 reshardCollection: "mydb.users",
5 key: { newShardKeyField: 1 }
6})
7
8// Check status of resharding operation
9db.adminCommand({ _shardsvrReshardingGetStatus: 1 })
10 A stale config exception occurs when a mongos router is working with outdated metadata about the chunk distribution.
This typically happens after a chunk migration completes but before all mongos routers refresh their metadata cache.
MongoDB automatically retries operations that encounter stale config exceptions, but persistent exceptions might
indicate issues with your config servers or network connectivity.
MongoDB sharding provides a powerful solution for horizontally scaling databases to handle large datasets and high
throughput requirements. Key takeaways from this lesson:
Sharding distributes data across multiple machines, enabling MongoDB to handle virtually unlimited data growth.
The choice of shard key is critical for performance, scalability, and query efficiency.
A sharded cluster consists of shard servers (holding data), config servers (holding metadata), and mongos routers (routing queries).
Ranged, hashed, and zoned sharding strategies offer different approaches to data distribution.
Effective monitoring and maintenance are essential for a healthy sharded cluster.
In the next lesson, we'll begin exploring Redis, a popular in-memory data structure store that provides
high-performance caching and serves as a versatile NoSQL solution for specific use cases.
Your progress is saved in this browser, on this device only.