Combining sharding with replica sets

Combining Sharding and Replica Sets

MongoDB's sharding and replica sets are two core high-availability solutions, designed for horizontal scaling and data redundancy, respectively. Sharding improves throughput by distributing data across multiple nodes, while replica sets ensure data safety through multiple copies. In real-world production environments, the two are often used together to achieve both scalability and disaster recovery capabilities.

Basic Architecture of a Sharded Cluster

A typical sharded cluster consists of the following components:

1. Shard: Each shard can be a standalone instance, but production environments recommend using replica sets as shards.
2. Config Server: Special mongod instances that store cluster metadata.
3. Query Router (mongos): The access point for applications.

// Example of connecting to mongos
const { MongoClient } = require('mongodb');
const uri = "mongodb://mongos1:27017,mongos2:27017/?replicaSet=rsMongos";
const client = new MongoClient(uri);

Implementation of Replica Sets as Shards

When using replica sets as shards, each shard is essentially a 3-node replica set:

shard1/
  ├── primary: shard1-node1:27018
  ├── secondary1: shard1-node2:27018 
  ├── secondary2: shard1-node3:27018
shard2/
  ├── primary: shard2-node1:27019
  ├── secondary1: shard2-node2:27019
  └── secondary2: shard2-node3:27019

In this architecture, even if the primary node of a shard fails, the replica set mechanism automatically elects a new primary node, ensuring continuous availability of the shard.

Choosing a Sharding Strategy

Range-Based Sharding

Suitable for queries with distinct range characteristics, such as time-series data:

sh.shardCollection("logs.events", { timestamp: 1 });

Hash-Based Sharding

A general solution to ensure even data distribution:

sh.shardCollection("users.profiles", { _id: "hashed" });

Tag-Based Sharding

Directs specific data to designated shards:

sh.addTagRange("orders.archive", 
  { _id: MinKey }, { _id: MaxKey }, "ARCHIVE");

Execution Flow of Read/Write Operations

Write Process

1. The application initiates a write via mongos.
2. The config server determines the target shard.
3. The write operation is routed to the primary node of the corresponding shard.
4. The primary node synchronizes the operation to secondary nodes.

Read Process

1. mongos parses the query conditions.
2. Determines the shards to access (may involve multiple shards).
3. Reads data from available nodes of each shard.
4. Merges results and returns them to the client.

// Set read preference to the nearest node
const options = {
  readPreference: 'nearest',
  maxStalenessSeconds: 30
};
const cursor = collection.find({}, options);

How the Balancer Works

The balancer is a background process in a sharded cluster responsible for:

• Monitoring data distribution across shards.
• Triggering chunk migrations when data is unevenly distributed.
• Ensuring normal service is unaffected during migrations.

Check the balancer status with:

use config
db.locks.find({ _id: "balancer" })

Failure Recovery Mechanisms

Shard Primary Node Failure

1. The replica set elects a new primary node within 30 seconds.
2. mongos automatically detects the topology change.
3. Subsequent requests are routed to the new primary node.

Entire Shard Unavailable

1. The cluster marks the shard as unavailable.
2. Queries targeting data on that shard return partial results.
3. The shard automatically resynchronizes upon recovery.

Key Monitoring Metrics

Focus on monitoring the following metrics:

• Data balance between shards.
• Chunk migration queue length.
• Replication lag of each shard's replica set.
• Connection pool usage of mongos.

// Example of checking shard status
db.adminCommand({ listShards: 1 });
// Check chunk distribution
db.chunks.find().pretty();

Performance Optimization Practices

Handling Hot Shards

When a shard is overloaded:

1. Check if the shard key selection is appropriate.
2. Consider adding new shards.
3. A temporary solution is to manually split hot chunks.

// Example of manually splitting a chunk
sh.splitAt("orders.transactions", { _id: "2023-06-01" });

Query Optimization Tips

1. Ensure query conditions include the shard key.
2. Avoid full-shard broadcast queries.
3. Use projections wisely to reduce data transfer.

// Good query practice
db.orders.find({
  orderDate: { $gte: ISODate("2023-01-01") },
  _id: customerId // _id is the shard key
});

Deployment Recommendations

For production environments, consider:

1. At least 3 config servers forming a replica set.
2. Each shard as a 3-node replica set.
3. Deploy multiple mongos instances for load balancing.
4. Start with 3 shards and scale as needed.

// Typical command to start mongos
mongos --configdb cfgReplSet/config1:27019,config2:27019,config3:27019 \
       --bind_ip_all

Capacity Planning Methods

Formula for calculating the required number of shards:

Total data volume × Growth factor × Replica count / Single-node capacity = Minimum shard count

Example:

• Expected data volume: 1TB
• Annual growth rate: 50%
• 3 replicas
• Single-node maximum: 500GB

(1000 × 1.5 × 3) / 500 = 9 shards

Handling Special Scenarios

Sharding Large Collections

For enabling sharding on existing collections:

1. First, create an index on the shard key.
2. Initial sharding may take a long time.
3. Perform during off-peak hours.

// Sharding an existing collection
db.adminCommand({
  shardCollection: "bigdata.records",
  key: { timestamp: 1, deviceId: 1 }
});

Cross-Shard Transactions

Version 4.0+ supports cross-shard transactions, but note:

• Significant performance overhead.
• Limit transaction scope and duration.
• May require adjusting transaction timeout settings.

// Example of a cross-shard transaction
const session = client.startSession();
session.startTransaction();
try {
  await orders.insertOne({...}, { session });
  await inventory.updateOne({...}, { session });
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
}