Overview
9 Using replication and sharding
This chapter explains how MongoDB achieves scale and resilience by combining replication and sharding. Replication copies data from a primary to secondaries to provide high availability, disaster recovery, and read scaling, while all writes target the primary. Sharding horizontally partitions collections so different shards host different slices of the data, distributing both read and write load; each shard should itself be a replica set for durability. Used together, they enable continuous availability, rolling upgrades, and large-scale growth. In managed environments, Atlas provisions replica sets by default and lets you add sharding for seamless horizontal scaling without heavy operational overhead.
The chapter details replica set mechanics: member roles and states, elections using protocol version 1 with terms, priorities, and timeouts, and driver behavior during failover. It explains the oplog as an ordered, idempotent stream that secondaries apply (including chained replication), with guidance on sizing, retention windows, monitoring lag, and initial sync options (logical and file-based). It introduces change streams for reliable, ordered, resumable change notifications across collections, databases, or deployments, with RBAC enforcement and aggregation pipeline filtering. Together, these features provide predictable failover, observable replication health, and real-time data events that applications can safely consume.
For sharding, the text covers cluster architecture (mongos routers, shards as replica sets, and config servers, including embedded config servers in 8.0), and the benefits of throughput, storage growth, and data locality via zones. It emphasizes shard key design—ranged vs hashed strategies, cardinality and monotonicity concerns, and the analyzeShardKey workflow with query sampling—plus practical operations: creating sharded clusters, detecting and correcting imbalance, chunk sizing and balancing windows, pre-splitting and moveChunk, handling jumbo chunks, and automatic chunk merging/defragmentation. It also describes lifecycle operations such as faster resharding (including forced redistribution), moving unsharded collections between shards, and unsharding back to a single shard. Finally, it shows how Read/Write Concern and Read Preference let you tune consistency, durability, and latency across replica sets and sharded clusters.
The election mechanism within a MongoDB replica set is activated when the primary node becomes non-operational. The secondary replicas conduct a vote to elect a new primary.
This MongoDB sharded cluster configuration includes multiple app servers connected to routers (mongos), which direct queries to the appropriate shard. Three shards, each a replica set, hold segments of the data. Additionally, there is an embedded config shard that contains the config servers (along with user data), which hold the cluster's metadata. Starting with MongoDB 8.0, config servers can be embedded in sharded clusters, simplifying the architecture and reducing overall infrastructure costs.
The two shards—atlas-up0q2r-config (config shard) and atlas-up0q2r-shard-0—are each a replica set with three nodes. In MongoDB 8.0+, a config server storing both metadata and application data is called a config shard. Both shards are located in the Iowa (us-central1) region. Key statistics, such as disk usage, operations, and connections over the last 6 hours, are displayed. Each node is labeled as PRIMARY or SECONDARY, indicating its replication status within the sharded cluster.
MongoDB 8.0 allows you to seamlessly move unsharded collections to a different shard.
Summary
- A replica set in MongoDB is a group of mongod processes that store identical data sets, ensuring redundancy and high availability. Members can exist in various states and perform different roles within the replica set.
- MongoDB uses protocol version 1, based on the Raft algorithm, for replica set elections, ensuring data consistency. It features a voting system to decide the primary member including the term concept. The oplog is a capped collection that logs ordered write operations in MongoDB, enabling replication. Write operations that don't change data or fail don't generate oplog entries.
- MongoDB change Streams offer reliable, ordered, and resumable change event notifications with role-based access control, ensuring data durability across replica sets. They also integrate with the aggregation pipeline for efficient server-side data processing, unlike oplog tailing.
- MongoDB uses sharding to partition data across multiple servers, enabling horizontal scaling and improved handling of large data volumes and high transaction rates. A MongoDB sharded cluster can consist of any number of shards, each holding a fragment of data, allowing the total database size to easily exceed 100 TB without a clear upper limit.
- MongoDB shards data at the collection level, distributing the collection data across the shards in the cluster using the shard key, which consists of one or more fields within the documents. Choosing the right shard key is crucial because an unsuitable selection can lead to inefficient data distribution, load imbalances across shards, and poor query performance.
- You can create a sharded cluster in MongoDB Atlas. Unlike the M0 cluster we set up in Chapter 1, these clusters are not free in Atlas, so it's necessary to set up a payment method. Sharded clusters are supported starting from the M30 tier.
- Starting with MongoDB 8.0, config servers can be embedded in sharded clusters, simplifying the architecture and reducing overall infrastructure costs.
- Choosing the right shard key is crucial in sharding clusters. A poor choice can lead to inefficient data distribution, load imbalances, and poor query performance. In some cases, an overloaded shard, known as a hot shard, can become a bottleneck, negatively impacting the cluster's overall performance.
- MongoDB simplifies shard key selection with the analyzeShardKey command, which provides metrics based on sampled queries for informed decision-making in both unsharded and sharded collections.
- MongoDB organizes sharded data into chunks or ranges, each defaulting to 128 MB in size. A chunk is defined by a shard key's lower bound (inclusive) and upper bound (exclusive), containing a continuous sequence of shard key values within a shard.
- Re-splitting chunks in a sharded collection can improve throughput, particularly when dealing with large data volumes in an unbalanced cluster or when ingesting data that could lead to imbalance, such as with monotonically changing shard keys.
- AutoMerger, integrated into the balancer's functionality, improves data shard management by automatically merging eligible data chunks based on predefined criteria, optimizing operations seamlessly in the background.
- In MongoDB 8.0, unsharded collections can be moved between shards without the need for additional configuration like sharding or selecting a shard key. This simplifies scaling horizontally and supports geographical distribution for compliance needs.
- In MongoDB 8.0, data resharding across shards is four times faster and includes support for time series collections, which reduces operational costs and simplifies database architecture management.
- In MongoDB 8.0, you can use the unshardCollection command to move all documents from a sharded collection onto a single shard, either a specified one or the shard with the least data.
- Write Concern defines how many replica set members must acknowledge a write operation for it to be successful. The levels include w: 1, where only the primary acknowledges; w: "majority", where a majority of members acknowledge; w: 0, where no acknowledgment is required; and w: "all", where all members must acknowledge.
- Read Concern defines the consistency level for read operations. The levels include local (reads from the current node), majority (data confirmed by the majority), linearizable (strict consistency), and snapshot (consistent view at a specific point in time).
- Read Preference determines which replica set members handle read operations. The options include primary (reads from the primary), primaryPreferred (primary if available, otherwise secondary), secondary (reads from secondaries), secondaryPreferred (secondary if available, otherwise primary), and nearest (reads from the node with the lowest latency).
FAQ
What’s the difference between replication and sharding in MongoDB?
- Replication: Copies data from a primary to secondaries for high availability and read scaling. All writes go to the primary, aiding disaster recovery but creating a potential write bottleneck.
- Sharding: Partitions a collection’s data across multiple shards (each shard should be a replica set) for horizontal scale and higher read/write throughput.
- Together: Use replication inside each shard to keep sharded clusters resilient and allow rolling upgrades.
- Atlas: Provides replication by default; sharding is an option for horizontal scaling and is fully managed.
Which replica set member states exist and how do I inspect them?
- Key states: PRIMARY (1), SECONDARY (2), RECOVERING (3), STARTUP (0), STARTUP2 (4), UNKNOWN (5), ARBITER (6), DOWN (7), ROLLBACK (8), REMOVED (9).
- Health/status:
db.adminCommand("replSetGetStatus")orrs.status(). - Config/details:
db.adminCommand("replSetGetConfig")orrs.config()to seepriority,votes,hidden,arbiterOnly,secondaryDelaySecs. - E.g. hidden/delayed members are useful for backups/analytics and protection against human error.
How do primary elections work and how long does failover take?
- Protocol PV1 (Raft-based): members vote; the most up-to-date node is favored; terms prevent double voting.
- Triggers: adding/removing nodes, init, or heartbeat failure (> ~10s by default).
- Duration: typically within ~12 seconds (detect + elect + finalize). Writes pause during election; reads can use secondaries if allowed by read preference.
- Tuning:
settings.electionTimeoutMillisin repl config. - Drivers auto-detect primary loss and retry certain ops.
What is the oplog, and how do I size and monitor it?
- Oplog: Capped collection (
local.oplog.rs) of idempotent logical writes. Primary streams ops to secondaries. - Inspect:
db.getReplicationInfo()(JSON, good for scripts),db.printSecondaryReplicationInfo()(human),rs.status(). - Defaults: WiredTiger ≈ 5% of disk (min ~990MB, max 50GB). In-Memory engine ≈ 5% of RAM (50MB–50GB).
- Resize: Before start via
--oplogSizeMB; after start viareplSetResizeOplog. - Retention window: Optionally enforce min hours with
storage.oplogMinRetentionHoursor--oplogMinRetentionHours. - Consider larger oplog for batch updates, high churn (deletes/inserts), or frequent in-place updates.
What are change streams and how do I start/resume them?
- Real-time, resumable change events for collections, databases, or deployments; respect RBAC; majority-committed; ordered (even across shards); support aggregation pipelines.
- Requirements: WiredTiger, replica set PV1.
- Start in mongosh:
db.collection.watch(pipeline)ordb.watch()/db.getSiblingDB("db").watch(). - Resume: Use the event’s
_idas resume token viaresumeAfterorstartAfter. Don’t modify_idin the pipeline. - Drivers: e.g., Node.js
collection.watch(pipeline)and handle reconnects per read preference.
How do I create and connect to a sharded cluster in MongoDB Atlas?
- Tier: Sharding is supported on M30+.
- Create (Atlas CLI):
atlas clusters create "MySharded" --provider GCP \ --region CENTRAL_US --tier M30 --type SHARDED --shards 2 --mdbVersion 8.0
- Load sample data:
atlas clusters sampleData load "MySharded". - Get SRV:
atlas clusters connectionStrings describe "MySharded". - Connect to mongos:
mongosh "mongodb+srv://<host>" --apiVersion 1 --username <user>
- Tip: Pause non-prod clusters to save costs.
How do I choose a good shard key and validate it?
- Strategies: Ranged (supports range queries) vs Hashed (evens writes for monotonic fields; not for range queries).
- Guidelines: Align with dominant query patterns; high cardinality; avoid skewed “hot” values; avoid monotonically increasing keys unless hashed.
- Analyze before sharding:
- Enable sampling:
db.coll.configureQueryAnalyzer({mode:"full", samplesPerSecond: N}) - Create supporting index for candidate key.
- Run workload; then
db.coll.analyzeShardKey({ ... })for metrics like single-shard reads%, distinct values, monotonicity.
- Enable sampling:
- Shard when satisfied:
sh.shardCollection("db.coll", { keySpec }).
How do I detect and correct data imbalance across shards?
- Detect:
- Cluster summary:
sh.status() - Per-collection:
db.coll.getShardDistribution() - All collections:
$shardedDataDistribution - Chunk stats:
db.getSiblingDB("config").chunks.aggregate([{ $group: { _id:"$shard", totalSize:{ $sum:"$size"}, count:{ $sum:1 }}}])
- Cluster summary:
- Balancer:
- Runs by data size; one migration per shard; up to n/2 parallel migrations.
- Window: Set via
use configthen updatedb.settings.updateOne({_id:"balancer"}, {$set:{activeWindow:{start:"HH:MM", stop:"HH:MM"}}}, {upsert:true}). - Control/Status:
sh.startBalancer(),sh.getBalancerState(),sh.isBalancerRunning().
- Manual fixes (use sparingly):
- Pre-split:
db.adminCommand({split:"db.coll", middle:{...}}) - Move chunk:
db.adminCommand({moveChunk:"db.coll", find:{...}, to:"shardX"}) - Jumbo: split with
sh.splitAt/findor clear flag viaclearJumboFlag. - AutoMerger (7.0+):
mergeAllChunksOnShard; control viash.startAutoMerger()/sh.enableAutoMerger().
- Pre-split:
How can I change data distribution later (reshard, move unsharded collections, unshard)?
- Reshard a sharded collection (5.0+; faster in 8.0):
db.adminCommand({ reshardCollection:"db.coll", key:{...} })- 8.0:
forceRedistribution:trueto redistribute under same key; parallelized reads/writes and bulk index build. - Plan for resources: ≥1.2× data size free disk, I/O < 50%, CPU < 80%; one collection at a time; avoid concurrent index builds/topology changes.
- 8.0:
- Move unsharded collection (8.0):
db.adminCommand({ moveCollection:"db.coll", toShard:"shardX" }) - Unshard a collection (8.0):
db.adminCommand({ unshardCollection:"db.coll", toShard:"shardX" })- Blocks writes ~2s; ensure capacity; one collection at a time; rebuild Atlas Search indexes afterward.
- Config servers (8.0): Transition between embedded/dedicated:
transitionFromDedicatedConfigServer,transitionToDedicatedConfigServer(use dedicated CSRS for >3 shards).
How do Read Concern, Write Concern, and Read Preference affect consistency and availability?
- Write Concern: Ack rules for durability (e.g.,
{w:"majority", j:true, wtimeout:5000}). - Read Concern: Consistency level (
local,available,majority,linearizable,snapshot). Example:db.coll.find(...).readConcern("majority"). - Read Preference: Target node for reads (
primary,primaryPreferred,secondary,secondaryPreferred,nearest). Example:db.coll.find(...).readPref("secondaryPreferred"). - Defaults: readConcern
local, writeConcernmajority. Tip: If you read-your-writes from primary, use readConcernmajoritywith writeConcernmajority. - Global defaults:
db.adminCommand({setDefaultRWConcern:1, defaultWriteConcern:{w:2}, defaultReadConcern:{level:"majority"}}).
What’s the practical workflow to shard a collection safely and efficiently?
- Profile workload and choose candidate keys (avoid hot/monotonic fields unless hashed).
- Create supporting index for the candidate key.
- Enable query sampling:
db.coll.configureQueryAnalyzer({mode:"full", samplesPerSecond:N}). - Run representative queries, then evaluate:
db.coll.analyzeShardKey({keySpec})(look for high single-shard read%, high distinct values, non-monotonicity). - Shard:
sh.shardCollection("db.coll", {keySpec}). - Monitor balance:
sh.status(),db.coll.getShardDistribution(), configure balancer window, and correct imbalances if needed.
MongoDB 8.0 in Action, Third Edition ebook for free
