Sharding: Horizontal Scaling

Sharded clusters are the ultimate solution for horizontal scaling in MongoDB—mastering them enables you to support petabyte-scale data.

1. What You'll Learn

• Sharding Concept (horizontal scaling)
• Shard Key Selection Strategies
• Chunk Splitting and Migration
• Range vs Hashed Sharding
• Config Server + Mongos + Shard Architecture
• Setting Up a Docker Sharded Cluster

2. Sharded Architecture

Concept Explanation: Sharding is MongoDB’s horizontal scaling solution—it distributes data across multiple shards, each of which is an independent replica set. Applications access the data transparently through Mongos routing, without needing to be aware of how the data is distributed. When a single server reaches its capacity or write throughput limit, sharding is the only way to scale.

How It Works: Mongos acts as a query router. After receiving a request from the application, it retrieves shard metadata (which chunks are on which shards) from the Config Server, routes the request to the target shard for execution, and merges the results before returning them. The application code is identical to that of a single-node MongoDB—sharding is transparent to the application.

Three Major Components:

graph TB
    App[Applications] --> M[Mongos<br/>Query Route<br/>Stateless]
    M --> CS[Config Server<br/>Configuration Information<br/>3Node Replica Set]
    M --> S1[Shard 1<br/>Shard Node 1<br/>Dungeon Collection]
    M --> S2[Shard 2<br/>Shard Node 2<br/>Dungeon Collection]
    M --> S3[Shard 3<br/>Shard Node 3<br/>Dungeon Collection]

    subgraph "Data Flow"
        Q[Query Request] --> M
        M -->|Metadata Query| CS
        M -->|Data Query| S1
        M -->|Data Query| S2
        M -->|Data Query| S3
    end

    style M fill:#cce5ff
    style CS fill:#fff3cd
    style S1 fill:#d4edda
    style S2 fill:#d4edda
    style S3 fill:#d4edda

graph TB
    App[Applications] --> M[Mongos<br/>Query Route<br/>Stateless]
    M --> CS[Config Server<br/>Configuration Information<br/>3Node Replica Set]
    M --> S1[Shard 1<br/>Shard Node 1<br/>Dungeon Collection]
    M --> S2[Shard 2<br/>Shard Node 2<br/>Dungeon Collection]
    M --> S3[Shard 3<br/>Shard Node 3<br/>Dungeon Collection]

    subgraph "Data Flow"
        Q[Query Request] --> M
        M -->|Metadata Query| CS
        M -->|Data Query| S1
        M -->|Data Query| S2
        M -->|Data Query| S3
    end

    style M fill:#cce5ff
    style CS fill:#fff3cd
    style S1 fill:#d4edda
    style S2 fill:#d4edda
    style S3 fill:#d4edda

Shards vs. Replica Sets:

3. Sharding Key Selection Strategies

Concept Explanation: A shard key is the field that determines how data is distributed across shards—it directly affects query performance, data balance, and scalability. Once set, a shard key cannot be modified; choosing the wrong shard key is one of the most serious architectural mistakes.

How It Works: MongoDB divides data into chunks (64 MB by default) based on the shard key, with each chunk assigned to a specific shard. Range sharding divides data based on ranges of shard key values, while hashed sharding divides data based on the hash values of the shard key. During a query, Mongos uses the shard key to locate the shard containing the target chunk, thereby avoiding broadcast queries.

ESRT Rules for Sharding Key Selection:

Four Characteristics of a Good Partition Key:

(1) Range Sharding (Range Partitioning)

Concept Explanation: Range sharding divides chunks based on the natural order of shard key values—consecutive value ranges are placed in the same chunk. This is suitable for range queries and sorting, but can easily lead to data skew (hotspots).

Principle: Range sharding divides the value range of the shard key into contiguous intervals [min, splitPoint1), [splitPoint1, splitPoint2), ..., with each interval corresponding to a chunk. Documents with adjacent values are placed in the same chunk → the same shard.

// === Enable Sharding ===
sh.enableSharding('shopdb');

// === Selecting a Sharding Key:User ID Scope ===
sh.shardCollection('shopdb.orders', { userId: 1 });
// userId 1-1000 → Shard 1
// userId 1001-2000 → Shard 2

(2) Hashed Shards

Concept Explanation: Hashed sharding calculates a hash for the shard key and divides chunks based on hash value ranges. Data is evenly distributed with no hotspots, but range queries require a broadcast to all shards.

// === Hash Sharding ===
sh.shardCollection('shopdb.products', { sku: 'hashed' });
// sku Hash values are evenly distributed across the shards

(3) Range vs Hashed Comparison

(4) Selecting the Best Sharding Key

// ✅ Good Sharding Key:High base figure、Infrequent Updates、Query hits
sh.shardCollection('shopdb.orders', { userId: 1, createdAt: -1 });
// Composite Segmented Key，userId Scope + Sort by Time

// ❌ Invalid Shard Key:Low base
sh.shardCollection('shopdb.products', { category: 1 });
// category Only 5 a value，It will tilt severely

Partition Key Anti-Pattern:

4. Chunk Splitting and Migration

Concept Explanation: A chunk is the smallest unit of sharded data management—64 MB by default—and contains all documents whose shard keys fall within a specific range. Chunks are automatically split when they exceed a threshold, and are automatically migrated when the number of chunks is uneven across shards. This is the core mechanism behind MongoDB’s automatic load balancing.

How It Works:

• Split: When a chunk exceeds 64 MB (or the number of documents exceeds the configured threshold), Mongos finds the median shard key and splits the chunk into two.
• Migration: The Balancer periodically checks the number of chunks in each shard and migrates chunks from shards with more chunks to those with fewer, until the load is balanced.

How an Equalizer Works:

graph TB
    subgraph "Equalizer Workflow"
        A[Check eachShard<br/>ChunkQuantity] --> B{Gap>8?}
        B -->|Yes| C[Select Migration Source Shard<br/>(Chunk with the most)]
        C --> D[Select a Migration DestinationShard<br/>(ChunkThe fewest)]
        D --> E[MigrationChunk]
        E --> F[UpdateConfig Server<br/>Metadata]
        F --> A
        B -->|No| G[Wait for the next check<br/>Default 10 seconds]
    end

    style E fill:#cce5ff
    style F fill:#d4edda

graph TB
    subgraph "Equalizer Workflow"
        A[Check eachShard<br/>ChunkQuantity] --> B{Gap>8?}
        B -->|Yes| C[Select Migration Source Shard<br/>(Chunk with the most)]
        C --> D[Select a Migration DestinationShard<br/>(ChunkThe fewest)]
        D --> E[MigrationChunk]
        E --> F[UpdateConfig Server<br/>Metadata]
        F --> A
        B -->|No| G[Wait for the next check<br/>Default 10 seconds]
    end

    style E fill:#cce5ff
    style F fill:#d4edda

graph LR
    A[Chunk 1<br/>1-1000] -->|Split| B[Chunk 1a<br/>1-500]
    A -->|Split| C[Chunk 1b<br/>501-1000]
    C -->|Migration| D[Shard 2]

    style B fill:#d4edda
    style D fill:#fff3cd

graph LR
    A[Chunk 1<br/>1-1000] -->|Split| B[Chunk 1a<br/>1-500]
    A -->|Split| C[Chunk 1b<br/>501-1000]
    C -->|Migration| D[Shard 2]

    style B fill:#d4edda
    style D fill:#fff3cd

Chunk Commands:

// === Manual Split Chunk ===
sh.splitAt('shopdb.orders', { userId: 5000 });

// === View Chunk Distribution ===
db.orders.getShardDistribution();

// === Balancer Status ===
sh.status();
sh.isBalancerRunning();

Key Points Analysis:

1. Chunk splitting is a logical operation (it only modifies metadata) and does not move data; chunk migration is a physical operation (it actually copies data).
2. The equalizer runs in the background and does not affect read/write operations during migration (using dual writes to ensure consistency).
3. During maintenance windows (such as batch imports), it is recommended to pause the equalizer to prevent the migration from affecting import performance.

▶ Example 1: Equalizer Management and Manual Chunk Splitting

// ShopHub:Pause the equalizer before importing a large batch，Restore After Import
// 1. Pause Equalizer
sh.stopBalancer();

// 2. Bulk Import Data
for (let i = 0; i < 1000000; i++) {
  db.orders.insertOne({ userId: 'user_' + (i % 500), total: Math.random() * 1000, createdAt: new Date() });
}

// 3. Manually Split Hotspots Chunk
sh.splitAt('shopdb.orders', { userId: 'user_100' });
sh.splitAt('shopdb.orders', { userId: 'user_200' });
sh.splitAt('shopdb.orders', { userId: 'user_300' });

// 4. Restore the Equalizer
sh.startBalancer();

// 5. Verification Distribution
db.orders.getShardDistribution();

5. Setting Up a Docker Sharded Cluster

Concept Overview: Deploying a sharded cluster is more complex than deploying a replica set—it requires a Config Server replica set, multiple shard replica sets, and Mongos routing. Docker Compose can orchestrate all these components with a single command, making it ideal for development and testing environments.

Deployment Architecture:

graph TB
    subgraph "Config Server Dungeon Collection"
        CS1[config1:27019]
    end

    subgraph "Shard 1 Dungeon Collection"
        S1A[shard1a:27018]
    end

    subgraph "Shard 2 Dungeon Collection"
        S2A[shard2a:27020]
    end

    subgraph "Mongos Routing"
        MS[mongos:27017]
    end

    App[Applications] --> MS
    MS --> CS1
    MS --> S1A
    MS --> S2A

    style MS fill:#cce5ff
    style CS1 fill:#fff3cd
    style S1A fill:#d4edda
    style S2A fill:#d4edda

graph TB
    subgraph "Config Server Dungeon Collection"
        CS1[config1:27019]
    end

    subgraph "Shard 1 Dungeon Collection"
        S1A[shard1a:27018]
    end

    subgraph "Shard 2 Dungeon Collection"
        S2A[shard2a:27020]
    end

    subgraph "Mongos Routing"
        MS[mongos:27017]
    end

    App[Applications] --> MS
    MS --> CS1
    MS --> S1A
    MS --> S2A

    style MS fill:#cce5ff
    style CS1 fill:#fff3cd
    style S1A fill:#d4edda
    style S2A fill:#d4edda

Comparison of Production vs. Development Configurations:

# docker-compose-sharding.yml
version: '3.8'
services:
  # Config Server(Dungeon Collection)
  config1:
    image: mongo:7.0
    command: mongod --configsvr --replSet configReplSet --port 27019

  # Shard 1(Dungeon Collection)
  shard1a:
    image: mongo:7.0
    command: mongod --shardsvr --replSet shard1ReplSet --port 27018

  # Mongos Routing
  mongos:
    image: mongo:7.0
    command: mongos --configdb configReplSet/config1:27019 --port 27017
    ports:
      - "27017:27017"

Initialization Steps:

# === Initialize a sharded cluster ===
# 1. Initialization Config Server Dungeon Collection
mongosh --port 27019 --eval 'rs.initiate({_id: "configReplSet", members: [{_id: 0, host: "config1:27019"}]})'

# 2. Initialization Shard 1 Dungeon Collection
mongosh --port 27018 --eval 'rs.initiate({_id: "shard1ReplSet", members: [{_id: 0, host: "shard1a:27018"}]})'

# 3. Through Mongos Add a shard
mongosh --port 27017 --eval '
  sh.addShard("shard1ReplSet/shard1a:27018");
  sh.enableSharding("shopdb");
  sh.shardCollection("shopdb.orders", { userId: 1 });
'

6. Sharded Cluster Monitoring

Concept Explanation: Monitoring a sharded cluster is more complex than monitoring a single node or a replica set—it requires monitoring the health of each component, data distribution balance, chunk migration status, and more. Proper monitoring can help detect data skew, hot shards, and configuration issues in a timely manner.

Monitoring Dimensions:

// === View Shard Status ===
sh.status();

// === Data Distribution ===
db.orders.getShardDistribution();

// === Balancer Management ===
sh.startBalancer();
sh.stopBalancer();
sh.setBalancerState(true);

// === Layout ===
db.settings.find();

Troubleshooting Common Issues:

▶ Example 2: Shard Monitoring and Troubleshooting

// Bob's DataFlow system has detected that queries are running slowly, diagnosing sharding issues
// 1. View Cluster Status
sh.status();
// Discovery Shard1: 8000 chunks, Shard2: 2000 chunks → Severe tilt

// 2. View Data Distribution
db.orders.getShardDistribution();
// Shard 1: 80GB (80%), Shard 2: 20GB (20%)

// 3. Check to see if there is Jumbo Chunk
db.config.chunks.find({ jumbo: true }).count();
// 3 Jumbo Chunks

// 4. Manually Split Hotspots Chunk
sh.splitAt('shopdb.orders', { userId: 'user_100' });
sh.splitAt('shopdb.orders', { userId: 'user_200' });

// 5. Verify that the equalizer is running
sh.startBalancer();
sh.isBalancerRunning(); // true

// 6. Waiting for the balancing process to complete，Check again
db.orders.getShardDistribution();
// Shard 1: 50GB (50%), Shard 2: 50GB (50%) → Balance

7. When Should You Use Sharding?

Concept Explanation: Sharding is not always best done as early as possible—while it provides scalability, it also increases operational complexity. Sharding too early can lead to unnecessary operational costs; sharding too late can result in single-server bottlenecks and difficulties with data migration. The decision should be based on a comprehensive assessment of data volume, write throughput, and growth trends.

Decision-Making Framework:

graph TB
    A{Data volume?} -->|< 100GB| B[❌ Single-player + Dungeon Collection]
    A -->|100GB-1TB| C{WriteQPS?}
    A -->|> 1TB| D[✅ Sharding]
    C -->|< 10K ops/s| E[⚠️ Dungeon Collection<br/>Monitor Growth Trends]
    C -->|> 10K ops/s| D

    B --> F[Optimize Indexes + Search]
    E --> G[Pre-planned Sharding Keys]
    D --> H[Select a partition key + Deploy a Sharded Cluster]

    style B fill:#d4edda
    style D fill:#cce5ff
    style E fill:#fff3cd

graph TB
    A{Data volume?} -->|< 100GB| B[❌ Single-player + Dungeon Collection]
    A -->|100GB-1TB| C{WriteQPS?}
    A -->|> 1TB| D[✅ Sharding]
    C -->|< 10K ops/s| E[⚠️ Dungeon Collection<br/>Monitor Growth Trends]
    C -->|> 10K ops/s| D

    B --> F[Optimize Indexes + Search]
    E --> G[Pre-planned Sharding Keys]
    D --> H[Select a partition key + Deploy a Sharded Cluster]

    style B fill:#d4edda
    style D fill:#cce5ff
    style E fill:#fff3cd

Pre-Sharding Checklist:

▶ Example: Hands-On Guide to Deploying a Sharded Cluster with Docker + Data Sharding

# === 1. Start the sharded cluster(docker-compose-sharding.yml)===
# services:
#   config1:
#     image: mongo:7.0
#     command: mongod --configsvr --replSet configReplSet --bind_ip_all --port 27019
#     ports: ["27019:27019"]
#
#   shard1a:
#     image: mongo:7.0
#     command: mongod --shardsvr --replSet shard1ReplSet --bind_ip_all --port 27018
#     ports: ["27018:27018"]
#
#   shard2a:
#     image: mongo:7.0
#     command: mongod --shardsvr --replSet shard2ReplSet --bind_ip_all --port 27020
#     ports: ["27020:27020"]
#
#   mongos:
#     image: mongo:7.0
#     command: mongos --configdb configReplSet/config1:27019 --bind_ip_all --port 27017
#     ports: ["27017:27017"]
#     depends_on: [config1, shard1a, shard2a]

docker-compose -f docker-compose-sharding.yml up -d

# === 2. Initialization Config Server Dungeon Collection ===
mongosh --port 27019 --eval '
  rs.initiate({
    _id: "configReplSet",
    members: [{ _id: 0, host: "config1:27019" }]
  });
'

# === 3. Initialization Shard Dungeon Collection ===
mongosh --port 27018 --eval '
  rs.initiate({ _id: "shard1ReplSet", members: [{ _id: 0, host: "shard1a:27018" }] });
'

mongosh --port 27020 --eval '
  rs.initiate({ _id: "shard2ReplSet", members: [{ _id: 0, host: "shard2a:27020" }] });
'

# === 4. Through Mongos Add a shard ===
mongosh --port 27017 --eval '
  sh.addShard("shard1ReplSet/shard1a:27018");
  sh.addShard("shard2ReplSet/shard2a:27020");
'

# === 5. Enable Database Sharding ===
mongosh --port 27017 --eval '
  sh.enableSharding("shopdb");
  sh.shardCollection("shopdb.orders", { userId: 1, createdAt: -1 });  // Composite Segmented Key
'

# === 6. Insert Test Data(Automatically distributed across different shards)===
mongosh --port 27017 --eval '
  for (let i = 0; i < 10000; i++) {
    db.orders.insertOne({
      userId: "user_" + (i % 100),
      total: Math.random() * 1000,
      createdAt: new Date(),
      status: "paid"
    });
  }
'

# === 7. View Shard Distribution ===
mongosh --port 27017 --eval 'sh.status();'
# Output:
# shards:
#   { _id: 'shard1ReplSet', count: 5012 }
#   { _id: 'shard2ReplSet', count: 4988 }
# The data is evenly distributed across 2 shards

# === 8. When querying Mongos Automatic Routing ===
mongosh --port 27017 --eval '
  db.orders.find({ userId: "user_50" }).count();
'
# Mongos Automatically route to the corresponding shard(Based on userId Scope)

// === 9. App Connections Mongos(App Transparency，Transparent Sharding)===
const mongoose = require('mongoose');
await mongoose.connect('mongodb://localhost:27017/shopdb');
// No changes to the application code are required，Compared to a standalone system MongoDB Exactly the same

await Order.create({
  userId: 'user_001',
  total: 599,
  items: [{ sku: 'PHONE-001', qty: 1 }]
});
// Mongos Automatically route to the appropriate shard

// === 10. Hashed Sharding(Uniform Distribution of Data)===
mongosh --port 27017 --eval '
  sh.shardCollection("shopdb.products", { sku: "hashed" });
'
// sku Uniform Distribution After Hashing，Avoid Hot Spots

Output: 10,000 orders are automatically distributed across 2 shards (5,012 + 4,988); the application connects via Mongos without needing to be aware of the sharding details.

❓ FAQ

📖 Summary

• Sharding Architecture: Mongos + Config Server + Shard
• Shard key selection: ESRT (Equality / Sort / Range / Time)
• Range vs Hashed Sharding
• Automatic Chunk Splitting and Balancing
• Setting Up a Docker Sharded Cluster
• When to partition: > 1 TB of data / > 10,000 ops/s

📝 Exercises

1. Basic Questions (⭐): Understand the sharding architecture and components (Mongo / Config Server / Shard).
2. Basic Question (⭐): Analyze your business scenario and determine whether sharding is necessary.
3. Advanced Exercise (⭐⭐): Deploy a cluster with 1 config and 2 shards using Docker Compose.
4. Advanced Question (⭐⭐): Test sharding key selection (userId vs. sku vs. compound).
5. Challenge (⭐⭐⭐): Deploy a complete sharded cluster (Config replica set + 2 shard replica sets + Mongos + monitoring).

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us