Replication and Sharding

Replication and sharding are two fundamental concepts in MongoDB that ensure data availability, fault tolerance, and the ability to scale your database as your application grows.

What is Replication?

Replication in MongoDB means keeping multiple copies of your data across different servers, called a replica set. A replica set consists of:

One primary node that handles all write operations
Multiple secondary nodes that automatically sync data from the primary

Think of it like having backup copies of important documents stored in different locations.

What is Sharding?

Sharding is MongoDB’s way of splitting large datasets across multiple machines. Instead of storing all data on one server, sharding divides your data into smaller pieces called shards and distributes them across multiple servers.

This is like organizing a huge library by putting different subjects on different floors.

Why Use Replication and Sharding?

These features provide three key benefits:

High Availability: If one server fails, others continue working
Better Performance: Distribute workload across multiple servers
Scalability: Handle growing data by adding more servers

Setting Up Replica Sets

Replica sets provide high availability by maintaining multiple copies of your data. Here’s how to set up a basic replica set.

Prerequisites

MongoDB installed on all servers
Network connectivity between servers
Sufficient disk space on each server

Basic Replica Set Setup

Start MongoDB instances with replica set name

On each server, start MongoDB with the --replSet parameter:

# Primary server (port 27017)
mongod --replSet "myReplicaSet" --port 27017 --dbpath /data/db1

# Secondary server 1 (port 27018)
mongod --replSet "myReplicaSet" --port 27018 --dbpath /data/db2

# Secondary server 2 (port 27019)
mongod --replSet "myReplicaSet" --port 27019 --dbpath /data/db3

Connect to the primary server
Terminal window
```
mongo --port 27017
```

Initialize the replica set

rs.initiate({
  _id: "myReplicaSet",
  members: [
    { _id: 0, host: "localhost:27017" },
    { _id: 1, host: "localhost:27018" },
    { _id: 2, host: "localhost:27019" },
  ],
});

Verify the setup
```
rs.status();
```

Managing Replica Set Members

Adding a new member
```
rs.add("localhost:27020");
```
Removing a member
```
rs.remove("localhost:27019");
```
Monitor replica set health
```
rs.printReplicationInfo();
```

Automatic Failover and Recovery

MongoDB replica sets provide automatic failover to ensure your application stays available even when servers fail.

How Failover Works

When the primary node fails, the remaining secondary nodes automatically:

Detect the primary is unreachable
Hold an election to choose a new primary
The new primary starts accepting write operations
Clients automatically reconnect to the new primary

What Happens During Recovery

When the original primary comes back online
- It rejoins as a secondary node
- Syncs any missed data from the current primary
- Continues serving read operations
When there are network issues
- Nodes that can communicate elect a new primary
- Once the network is restored, all nodes resync
- Data consistency is maintained across all nodes

Understanding Sharding

Sharding allows MongoDB to distribute data across multiple servers, enabling your database to handle larger datasets and higher throughput.

Sharding Components

Shards
- Each shard stores a portion of your data
- Usually implemented as replica sets for redundancy
- Can be added or removed as needed
Config Servers
- Store metadata about which data lives on which shard
- Track the mapping of data chunks to shards
- Usually deployed as a 3-member replica set
Router (mongos)
- Routes client requests to the correct shard
- Merges results from multiple shards
- Acts as a gateway between your application and the sharded cluster

How Data Gets Distributed

MongoDB uses a shard key to determine which shard stores each document. The shard key is a field (or combination of fields) that exists in every document in your collection.

For example, if you use userId as your shard key:

Documents with userId: 1-1000 might go to Shard A
Documents with userId: 1001-2000 might go to Shard B
And so on…

Choosing Good Shard Keys

High Cardinality

Choose a shard key with many unique values to distribute data evenly:

// Good: User ID (many unique values)
{
  userId: ObjectId("...");
}

// Bad: Status (only a few values like "active", "inactive")
{
  status: "active";
}

Match Your Query Patterns

Consider how your application queries data:

// If you often query by user:
db.orders.find({ userId: "12345" });

// Use userId as shard key
sh.shardCollection("mydb.orders", { userId: 1 });

Avoid Hotspots

Don’t use fields that create uneven data distribution:

// Bad: Timestamp (new data always goes to one shard)
{ createdAt: ISODate("...") }

// Better: Combine timestamp with user ID
{ userId: 1, createdAt: 1 }

Consider Write Patterns

Make sure writes are distributed across shards, not concentrated on one.

Setting Up a Sharded Cluster

Start Config Servers

# Start 3 config servers
mongod --configsvr --replSet configReplSet --port 27019 --dbpath /data/configdb1
mongod --configsvr --replSet configReplSet --port 27020 --dbpath /data/configdb2
mongod --configsvr --replSet configReplSet --port 27021 --dbpath /data/configdb3

Start Shard Servers

# Shard 1
mongod --shardsvr --replSet shard1 --port 27018 --dbpath /data/shard1

# Shard 2
mongod --shardsvr --replSet shard2 --port 27019 --dbpath /data/shard2

Start Router (mongos)

mongos --configdb configReplSet/localhost:27019,localhost:27020,localhost:27021 --port 27017

Add Shards to Cluster

Connect to mongos and add shards:

sh.addShard("shard1/localhost:27018");
sh.addShard("shard2/localhost:27019");

Enable Sharding

// Enable sharding for database
sh.enableSharding("myDatabase");

// Shard a collection
sh.shardCollection("myDatabase.orders", { userId: 1 });

Managing Sharded Clusters

Monitor Cluster Health

// Check cluster status
sh.status();

// Check balancer status
sh.getBalancerState();

// View chunk distribution
sh.status();

Balancing Data

MongoDB automatically balances data across shards, but you can control this:

// Start the balancer
sh.startBalancer();

// Stop the balancer
sh.stopBalancer();

// Check if balancer is running
sh.isBalancerRunning();

Adding New Shards

As your data grows, add more shards:

// Add a new shard
sh.addShard("shard3/localhost:27020");

// MongoDB will automatically rebalance data

Best Practices

For Replica Sets

Use odd numbers of members (3, 5, 7) to avoid election ties
Deploy members across different data centers for better availability
Monitor replica lag regularly
Use appropriate read preferences for your use case

For Sharding

Choose shard keys carefully - they can’t be changed easily
Start with fewer shards and add more as needed
Monitor chunk distribution and balance regularly
Consider the impact of shard key on your queries

General Tips

Test failover scenarios in development
Monitor performance metrics regularly
Have proper backup and recovery procedures
Plan for capacity growth

Common Issues and Solutions

Replica Set Issues

Problem: Replica set members can’t reach each other Solution: Check network connectivity and firewall settings

Problem: Primary election takes too long Solution: Reduce election timeout settings or check network latency

Sharding Issues

Problem: Uneven data distribution Solution: Review shard key choice and consider pre-splitting chunks

Problem: Poor query performance Solution: Ensure queries include the shard key when possible

Conclusion

Replication and sharding are powerful features that help your MongoDB deployment handle growth and maintain availability. Start with replica sets for high availability, then add sharding when you need to scale beyond a single machine’s capacity.

Remember to plan your architecture carefully, monitor performance regularly, and test failure scenarios to ensure your setup meets your application’s needs.