Skip to content
Deadline to Die

Sharding Architecture

Sharding Architecture

Sharding is the process of distributing data across multiple machines. It is MongoDB’s approach to meeting the demands of massive data growth and high-throughput applications.

🏗️ 1. Sharded Cluster Components

A MongoDB sharded cluster consists of three main components, each playing a critical role in data distribution and retrieval.

Shards

A shard is a subset of the cluster’s data.

  • Each shard can be an individual mongod instance, but in production, each shard MUST be a Replica Set.
  • Shards provide high availability and data redundancy.

Config Servers

Config servers store the cluster’s metadata and configuration settings.

  • They map specific data ranges to the shards that hold them.
  • Like shards, config servers must be deployed as a Replica Set.

Mongos (Query Routers)

mongos instances are the interface between the client application and the cluster.

  • They route operations to the appropriate shard(s) based on the shard key.
  • They aggregate results from multiple shards and return them to the client.

🚀 2. Data Distribution Strategy

Data is split into Chunks (default size: 64MB) based on the Shard Key.

Range-Based Sharding

Data is partitioned into ranges based on the shard key’s value.

  • Pros: Efficient for range-based queries (e.g., timestamp ranges).
  • Cons: Can lead to “hotspots” if writes are monotonically increasing (e.g., all new logs hitting the same shard).

Hash-Based Sharding

A hash of the shard key’s value is used to distribute data.

  • Pros: Ensures an even distribution of data across all shards, preventing hotspots.
  • Cons: Inefficient for range-based queries (requires scanning all shards).

⚡ 3. Cluster Operations

Pymongo Example: Enabling Sharding

from pymongo import MongoClient

client = MongoClient('mongodb://mongos:27017/')
admin_db = client['admin']

# Step 1: Enable sharding for the database
admin_db.command("enableSharding", "analytics")

# Step 2: Shard a specific collection using a hashed key
admin_db.command("shardCollection", "analytics.logs", key={"_id": "hashed"})

Mongo Shell Example: Checking Cluster Status

// Check the status of the sharded cluster
sh.status();

// View current shards
db.adminCommand({ listShards: 1 });

🛡️ 4. When to Shard?

Sharding adds significant operational complexity. You should only consider sharding when:

  1. Storage Limits: Your data volume exceeds the storage capacity of a single node.
  2. Throughput Limits: Your write operations exceed the capacity of a single Primary node.
  3. Memory Limits: Your working set (active data + indexes) no longer fits in the RAM of a single server.

💡 Best Practices

  1. Dedicated Infrastructure: Run mongos and Config Servers on dedicated machines to avoid resource contention.
  2. Monitor Chunk Distribution: Use sh.status() to ensure the balancer is successfully moving chunks across shards.
  3. Choose the Shard Key Wisely: The choice of shard key is the most critical decision and is difficult to change later.