Lesson 40 of 50 – MongoDB Sharding
80%

MongoDB Sharding

MongoDB sharding is a technique used to distribute data across multiple servers. It helps MongoDB handle large datasets and high-throughput workloads by distributing storage and processing across multiple machines.

1. What Is Sharding?

Imagine a school management application has millions of student records. Storing and processing all the data on one server may eventually become expensive or insufficient.

Sharding distributes a collection's documents across multiple servers called shards.

             MongoDB Sharded Cluster
                       |
                  mongos Router
                       |
             ---------------------
             |         |         |
           Shard 1   Shard 2   Shard 3
             |         |         |
          Some data  Some data  Some data

Each shard stores a portion of the sharded data. The router directs queries to the appropriate shard or shards.

2. Why Is Sharding Needed?

  • Large datasets: Distributes data across multiple servers.
  • Horizontal scaling: Allows capacity to grow by adding suitable servers.
  • Workload distribution: Can distribute reads and writes across shards.
  • Storage capacity: Supports datasets that exceed the practical capacity of one server.
  • Scalability: Helps applications grow as data and traffic increase.

Sharding introduces additional operational complexity. It is generally useful when a single server or replica set cannot adequately meet the application's storage or workload requirements.

3. Main Components of a Sharded Cluster

Component Purpose
Shard Stores a portion of the sharded data.
mongos Routes client requests to the appropriate shards.
Config servers Store cluster metadata and sharding configuration.
Replica set Commonly used to provide redundancy for each shard and for the config servers.

Production sharded clusters normally use replica sets for the shards and config servers. The mongos router directs operations using cluster metadata maintained by the config servers.

4. What Is a Shard Key?

A shard key is a field or combination of fields used to distribute documents across shards.

Example student document:

{
    "_id": ObjectId("..."),
    "student_id": 1001,
    "name": "Rahul Kumar",
    "city": "Aurangabad",
    "course": "Computer Science"
}

A possible shard key could be { student_id: 1 }, depending on the application's query patterns and data distribution.

The shard key is an important design decision. A poor choice can create uneven data distribution or concentrate writes on a small part of the cluster.

5. Types of Shard Keys

1. Ranged Sharding

Ranged sharding distributes documents according to ranges of shard-key values.

{ student_id: 1 }

This can support queries that search for ranges of student IDs. However, sequentially increasing values can cause uneven write distribution in some workloads.


2. Hashed Sharding

Hashed sharding uses a hashed value of the shard-key field to distribute documents.

{ student_id: "hashed" }

Hashing can help distribute monotonically increasing values more evenly. However, range queries on the original field may need to access multiple shards.


3. Compound Shard Key

A compound shard key uses multiple fields.

{ city: 1, student_id: 1 }

The order of fields matters. Choose the key based on actual queries, cardinality, distribution, and write patterns rather than simply selecting fields at random.

6. Sharding vs Replication

Sharding Replication
Distributes data across shards. Maintains copies of a data set across replica set members.
Primarily supports horizontal scaling. Primarily supports redundancy and availability.
Each shard stores a portion of the sharded collection. Data-bearing members of a replica set maintain the same data set.
Can use multiple replica sets as shards. Can be used independently without sharding.

A sharded cluster commonly uses replication too. Sharding and replication solve different problems and can work together.

7. Enable Sharding for a Database

The following commands illustrate the shell operations used with an already configured and running sharded cluster. They do not create a complete cluster by themselves.

Connect to a mongos router:

mongosh "mongodb://localhost:27017"

In this example, port 27017 represents the address of the router, not necessarily a shard server.

Enable sharding for the database:

sh.enableSharding("school")

This enables sharding for the school database. The database does not automatically have every collection sharded; collections must be configured separately.

8. Shard a Collection

After enabling sharding for the database, choose a suitable shard key for the collection.

sh.shardCollection(
    "school.students",
    { student_id: 1 }
)

This command requests ranged sharding for the school.students collection using student_id as the shard key.

The command assumes that the cluster is correctly configured and that the chosen shard key satisfies the applicable MongoDB requirements. If the collection already contains data, review the shard-key requirements and indexing rules for your MongoDB version.

Important: Do not run sharding commands against a production database without reviewing the shard key, existing indexes, cluster configuration, and migration impact first.

9. Insert and Query Data

Once the collection is sharded, applications can continue using standard MongoDB CRUD operations through the mongos router.

Insert sample documents:

use school

db.students.insertMany([
    {
        student_id: 1001,
        name: "Rahul Kumar",
        city: "Aurangabad"
    },
    {
        student_id: 1002,
        name: "Priya Singh",
        city: "Patna"
    },
    {
        student_id: 1003,
        name: "Amit Kumar",
        city: "Gaya"
    }
])

Query using the shard key:

db.students.find({
    student_id: 1001
})

Including the shard key in a query can help MongoDB target the relevant shard instead of broadcasting the query to every shard.

The actual routing behavior depends on the shard key, the query, and the cluster's metadata.

10. Check Sharding Information

View the cluster's sharding status:

sh.status()

This displays information about shards, databases, sharded collections, and their shard-key configuration.

You can also inspect collection statistics:

db.students.stats()

Review the output to understand the collection's statistics and sharding-related information exposed by your MongoDB version.

11. Common Sharding Problems

  • Uneven distribution: Some shards store or process much more data than others.
  • Poor shard key: The key does not match the application's query or write patterns.
  • Scatter-gather queries: Queries without useful shard-key constraints may need to reach multiple shards.
  • Configuration problems: Shards, config servers, or routers may not be reachable.
  • Migration overhead: Moving data between shards consumes network and server resources.
  • Insufficient capacity: Adding shards does not automatically fix every bottleneck.

12. Best Practices

  • Evaluate whether a replica set alone meets your requirements before adopting sharding.
  • Choose a shard key based on query patterns and distribution.
  • Monitor shard balance, query performance, and resource usage.
  • Include shard-key fields in queries where appropriate.
  • Secure cluster communication and database access.
  • Maintain independent backups and test recovery procedures.
  • Test the expected workload before deploying changes to production.

13. Practical Assignment

  1. Explain sharding in your own words.
  2. Draw a diagram showing shards, a router, and config servers.
  3. Explain the purpose of a shard key.
  4. Compare ranged and hashed sharding.
  5. Describe the difference between replication and sharding.
  6. On a suitable test cluster, enable sharding for the school database.
  7. Shard a test collection using a suitable shard key.
  8. Insert and query sample documents through the router.
  9. Inspect the cluster using sh.status().
  10. Explain why shard-key selection affects performance.

🧠 Quick Quiz

1. What is the main purpose of sharding?




2. What does a shard key help determine?




3. Which component routes client requests in a sharded cluster?