0% found this document useful (0 votes)
8 views51 pages

Understanding MongoDB Sharding Basics

Uploaded by

pks.08022004
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
8 views51 pages

Understanding MongoDB Sharding Basics

Uploaded by

pks.08022004
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

29-11-2025

Sharding
Unit – 4
Chapter -1

Content:
• Sharding: Introduction to Sharding Understanding the Components
of a Cluster; A One-Minute Test Setup.

1
29-11-2025

• Sharding in NoSQL is a technique for distributing a single dataset


across multiple servers by splitting it into smaller, independent
"shards".
• This is a core feature of NoSQL databases that enables horizontal
scalability, improved performance, and higher availability by
spreading the data and workload across a cluster.
• NoSQL databases often provide automatic sharding, using a shard
key (like a user ID) to determine where data is stored, which
simplifies scaling for large datasets.

How it works

1. Partitioning: The database divides a large dataset into smaller, more


manageable chunks called shards.

2. Distribution: Each shard is stored on a separate server, or node, in the


cluster.

3. Shard Key: A shard key, which can be a field like a user ID or geographic
location, determines which shard a piece of data belongs to.

4. Query Routing: When a query is made, a router directs the request to the
correct shard(s) for processing.

[Link]: As the data grows, the database can automatically


rebalance the data by creating new shards and redistributing the load.

2
29-11-2025

Benefits

Scalability: Allows databases to handle massive amounts of data and


high traffic by scaling out with more commodity hardware, instead of
scaling up a single machine.

Performance: Distributes the workload across multiple servers,


leading to faster reads and writes because each server only has to
manage a subset of the total data.

Availability: Improves fault tolerance, as the failure of a single server


only affects the data on that specific shard, not the entire database.

Examples in NoSQL databases


• MongoDB: Uses automatic sharding to distribute data across
different servers, with the shard key dictating the distribution.
• Cassandra: Distributes data by hashing the primary key across a
set of nodes.

3
29-11-2025

Understanding the Components of a Cluster


• Sharding in MongoDB lets you split your data across multiple
machines (called shards) so your database can handle more data
and traffic than a single server can.

• Many people confuse replication and sharding:


• Replication makes copies of the same data on multiple servers
(each one is identical).
• Sharding splits the data into parts, and each shard stores a
different subset of the data.

Understanding the Components of a Cluster


• One main goal of sharding is to make a group of many servers (like 5, 10,
or even 1,000) act as one single server to your application.
• To achieve this, MongoDB uses a special routing service called mongos.
• The mongos router acts as a middle layer between the application and
the shards.
• It maintains a kind of “table of contents” that tells it which shard
contains which portion of the data.
• When your application sends a request (for example, to read, write, or
update data), the mongos router receives it first.
• Then it checks its internal table to determine which shard or shards hold
the required data.

4
29-11-2025

Understanding the Components of a Cluster


After that:
• mongos forwards the request to the correct shard(s).
• Each shard processes its part of the request.
• mongos then collects and merges the responses from all involved shards.
• Finally, it sends the combined result back to the application.

• All the complexity of data distribution, request forwarding, and result merging is
handled automatically by the mongos router.

• This design makes it easy to scale MongoDB horizontally — meaning you can
add more machines (shards) to handle more data or users, without changing
how your application connects to the database.

5
29-11-2025

Example: Online Shopping Database

• Imagine you have a database for an online shopping app with 1 crore
(10 million) customer records.
If you store all that data on one server, it will become slow and
overloaded.
So, you split the data into parts and store each part on a different
shard.

Shard Data Stored

Shard 1 Customers whose names start with A–H

Shard 2 Customers whose names start with I–P

Shard 3 Customers whose names start with Q–Z

• When a user named “Priya” logs in:


• The app sends the request to mongos (router).
• Mongos checks where “Priya”’s data is stored (Shard 2).
• It forwards the request only to Shard 2.
• Shard 2 returns the data to the router, which sends it back to the app.

In short:
Each shard holds a subset of the total data.
Together, all shards make up the complete database, allowing MongoDB
to handle large-scale data efficiently.

6
29-11-2025

A One-Minute Test Setup


• To understand how sharding works in MongoDB, you can create a
small test cluster on a single machine.

Step 1: Start Mongo shell


• First, open a mongo shell without connecting to any database by
using the --nodb option:
$ mongo –nodb

• This opens the shell where you can manually create a cluster setup.

Step 2: Create a Sharded Cluster


• In the Mongo shell, use the ShardingTest class to create a test cluster:
cluster = new ShardingTest({"shards": 3, "chunksize": 1})
• Here:
• "shards": 3 means the cluster will have three shards (three mongod processes).
• "chunksize": 1 is a configuration setting for how MongoDB splits data into “chunks.”
When you run this command:
• MongoDB automatically starts three shard servers on ports 30000,
30001, and 30002.
• It also starts a mongos router on port 30999.
• The mongos acts as the front-end router for your cluster—it’s what your
application (or shell) will connect to.

7
29-11-2025

Step 3: Connect to the Cluster


• Because the main shell is showing all cluster logs, it’s better to open
another shell to connect to the cluster.
In the second shell, connect to the mongos router like this:
db = (new Mongo("localhost:30999")).getDB("test")

• Now you’re connected to the mongos on port 30999, using the test
database.

Step 4: How it Works


• At this point, your shell is acting as the client, and the mongos
router is your connection point to the sharded cluster.
• You can now send database commands (like inserts, queries, or
updates) to the mongos.
• The mongos will automatically:
• Figure out which shard(s) hold the relevant data.
• Forward the requests to the correct shards.
• Collect and merge the results.
• Send the final response back to your shell.

8
29-11-2025

Step-by-step Example
1. Start mongosh without connecting to any database:
mongo –nodb

2. Create a test cluster with 3 shards:


cluster = new ShardingTest({ shards: 3, chunksize: 1 })
• shards: 3 → creates three shard servers
• chunksize: 1 → keeps chunk size small for testing

3. Connect to the cluster’s mongos:


db = [Link]("testDB")

4. Create a sample collection and insert some data:


for (let i = 1; i <= 10; i++) {
[Link]({ student_id: i, name: "Student" + i, marks: i * 10 })
}

9
29-11-2025

5. Enable sharding for the database:


[Link]("testDB")

6. Shard the collection with a shard key:


• Here, we’ll use student_id as the shard key.
[Link]("[Link]", { student_id: 1 })

• { student_id: 1 } → means ascending order on the field student_id.

7. Check the sharding status:


[Link]()

• You’ll see:
• [Link] is sharded.
• Shards like shard0000, shard0001, etc.
• Distribution of chunks across shards.

8. View which shard holds which documents:


[Link]()

10
29-11-2025

Shard Configuration
Chap -2
Unit - 4

Starting the Servers


• Three main components of the cluster:
• Config servers — store metadata about the cluster (which shard has
which data, which collections are sharded, etc).

• Shard servers — these actually hold the data (each shard is a replica
set).

• mongos (router) — a process that clients connect to; it routes


queries to the correct shard based on the metadata.

1
29-11-2025

Introduction
• In MongoDB, config servers are essential components of a sharded
cluster, responsible for storing the cluster's metadata.

• This metadata includes information about chunks, their ranges, and


their distribution across shards.
• Mongos instances use this metadata to route read and write
operations to the correct shards.

Config Servers
• Config servers are the brains of your cluster: they hold all of the metadata
about which servers hold what data.
• Set up config server first .
• Make sure that they are running with journaling enabled and that
• Their data is stored on non-ephemeral drives.
(Ensure journaling is turned on, so writes are safely logged on disk. Use
persistent storage, not temporary (ephemeral) drives, so the data is not
lost if the server restarts).

2
29-11-2025

MongoDB Sharding Setup


• A sharded cluster in MongoDB has three key components:

• Config Servers (--configsvr) — store metadata (cluster config, chunk


info, etc.)

• Shard Servers (--shardsvr) — store the actual data (each is typically a


replica set)

• Mongos Router (mongos) — routes queries to the correct shard.

• A MongoDB sharded cluster uses shards, config servers, and query


routers to ensure efficient data distribution, high availability, and
scalability.
• Here are the steps to setting up MongoDB Sharded cluster:

3
29-11-2025

Step 1: Start the Config Servers


• Config servers store the cluster’s metadata. Start three config servers on different
machines:
• Each config server is a standalone mongod process started with --configsvr.

# Config Server 1
mongod --configsvr --replSet configReplSet --port 26050 --dbpath /data/config1 --
bind_ip localhost

# Config Server 2
mongod --configsvr --replSet configReplSet --port 26051 --dbpath /data/config2 --
bind_ip localhost

# Config Server 3
mongod --configsvr --replSet configReplSet --port 26052 --dbpath /data/config3 --
bind_ip localhost

Repeat this on all three config servers. Then, initiate the replica set by
connecting to one of the servers:
mongo --port 27019
Run:
[Link]({
_id: "configReplSet",
configsvr: true,
members: [
{ _id: 0, host: "config1:27019" },
{ _id: 1, host: "config2:27020" },
{ _id: 2, host: "config3:27021" }
]
})

4
29-11-2025

Start Shard Servers


• Each shard should be a replica set. Start three shard nodes on different
[Link] for three shards:

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

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

# Shard 3
mongod --shardsvr --replSet shard3 --port 27020 --dbpath /data/shard3 --
bind_ip localhost

Repeat this on all shard servers. Then, initiate the replica set:
mongo --port 27018
• Run:
[Link]({
_id: "shardReplSet1",
members: [
{ _id: 0, host: "shard1:27018" },
{ _id: 1, host: "shard2:27019" },
{ _id: 2, host: "shard3:27020" }
]
})

5
29-11-2025

Start the Mongos Router


• Mongos connects the client to the cluster via the config server replica
set.
• mongos --configdb
configReplSet/localhost:26050,localhost:26051,localhost:26052 --port
27017 --bind_ip localhost

Connect to Mongos and Add Shards


Now connect to Mongos:
mongo --port 27017

Then add shards to the cluster:


[Link]("shard1/localhost:27018")
[Link]("shard2/localhost:27019")
[Link]("shard3/localhost:27020")

6
29-11-2025

Enable Sharding for a Database


• Once shards are added:
[Link]("school")

• Then, shard a collection on a key:


[Link]("[Link]", { student_id: 1 })

Component Command Example Role

Config Server mongod --configsvr ... Stores metadata

Shard Server mongod --shardsvr ... Stores actual data

Mongos mongos --configdb ... Query router

7
29-11-2025

Adding Capacity to a Sharded Cluster


• When your data grows and you need more storage or performance
capacity, you can expand your cluster by adding new shards.

1. Adding a New Empty Shard


• To add a new shard:
• Create a new replica set for the shard.
• Make sure it has a unique replica set name, different from existing shards.
• Initialize the replica set and ensure that a primary node is elected.
[Link]("replicaSetName/host1:port,host2:port,host3:port
")
• This command adds the new replica set as a shard in the cluster

• Adding Existing Replica Sets as Shards


• If you already have several independent replica sets that are not yet
part of a cluster, you can also add them as shards — provided they do
not contain databases with the same names.
• For example:
Replica set A has the blog database
Replica set B has the calendar database
Replica set C has the mail and music databases
• You can safely add all three as shards — your cluster will then contain
three shards and four databases.
• However, if another replica set D also contains a database named
mail, mongos will reject it to prevent conflicts between duplicate
database names across shards.

8
29-11-2025

Task Description
Add capacity Add new shards (replica sets)
Requirement Each shard must have unique replica set name
Condition No overlapping database names across shards
Command [Link]("replicaSetName/host:port")

Sharding Data in MongoDB


• MongoDB does not automatically distribute data across shards.
You must explicitly configure which databases and collections are to
be sharded and specify how they should be partitioned.

• Enabling Sharding for a Database


• Before you can shard a collection, you must first enable sharding on
its parent database.
• For example, to enable sharding on the music database:
[Link]("music")
Note: Sharding must always be enabled at the database level before
you can shard any of its collections.

9
29-11-2025

Sharding a Collection
• Once sharding is enabled on the database, you can shard a collection
by running the [Link] command and specifying a shard
key.
Example:
[Link]("[Link]", { "name": 1 })
• This command shards the artists collection in the music database
using the "name" field as the shard key.
The { "name": 1 } syntax specifies ascending order for the key.

Index Requirement
• If the collection already exists, there must be an index on the shard
key ("name" in this case).
If the index does not exist, MongoDB will return an error and even
suggest the index to create.
• You can fix it by creating the required index:
[Link]({ "name": 1 })
• Then retry the [Link] command.
• If the collection does not yet exist, MongoDB automatically creates the
shard key index for you.

10
29-11-2025

Step Command Description


Enable database Allows collections in
[Link]("music")
sharding music to be sharded
[Link]("[Link]", { Shards data by the
Shard a collection
"name": 1 }) "name" field
Must exist before
Index requirement [Link]({ "name": 1 }) sharding existing
collection
Enables horizontal
Result Data split into chunks and balanced across shards
scaling

How MongoDB Tracks Cluster Data


• In a sharded cluster, every mongos router must know where to find a
document, given its shard key.

• MongoDB organizes data into chunks.

• What Is a Chunk?
• A chunk is a contiguous range of shard key values.

• Each chunk contains all documents whose shard key values fall within a
specific range, and each chunk resides entirely on one shard.

• MongoDB maintains a mapping table (stored on the config servers) that


associates:
<shard key range> → <shard>
• This allows mongos to determine which shard holds a given document or
query range.

11
29-11-2025

Chunk Range Stored On

Chunk Mapping by Shard Key age: 0–17 Shard A


• Suppose we have a collection
users sharded on { "age": 1 }.
[Link]({ age: 5 }) age: 18–40 Shard B
• It can instantly route that query
to Shard A, because the value 5
falls in the range 0–17. age: 41–70 Shard C

• Chunk Splitting
• As new documents are inserted, chunks can grow in size.
If a chunk becomes too large, MongoDB will automatically split it
into two smaller chunks to maintain balance.
• For example:
Original chunk: age 3–17
• After splitting:
• Chunk 1: age 3–11
• Chunk 2: age 12–17
• These two chunks together still cover the entire original range (3–17),
but are smaller and easier to distribute across shards.

12
29-11-2025

Concept Description

Chunk Logical range of shard key values

One chunk per shard Each chunk resides fully on a single shard

Chunk splitting Automatic when a chunk grows too large

No overlaps Prevents duplicate data ranges

No array shard keys Because arrays map to multiple index entries

Chunks are logical Not tied to physical disk layout

Splitting Chunks
• A chunk is a range of shard key values that represents a portion
of data within a collection.
• Each shard holds several chunks.
• MongoDB tries to keep chunk sizes roughly equal to ensure
balanced distribution of data and load.

13
29-11-2025

1. Tracking Inserts
• The mongos (the query router) keeps track of how much data it
inserts into each chunk.
• Each chunk has an internal counter of how many inserts or how
much data has been added.

2. Checking Split Threshold


• When the data added to a chunk reaches a threshold (for example,
64 MB by default), mongos checks whether the chunk needs to be
split.

14
29-11-2025

3. Chunk Split Decision


• If splitting is needed:
• The metadata on the config servers (which track all chunks and their
ranges) is updated.
• This involves:
• Creating new chunk documents to represent the new chunks.
• Modifying the “max” range of the old chunk to mark where it ends.
• Important: No actual data is moved yet — it’s only a metadata change.

4. Reset Tracking
• After the split:
• The mongos resets its tracking counters for the original chunk.
• New trackers are created for the newly created chunks.

5. Shard’s Role
• When mongos asks the shard whether a chunk should be split:
• The shard estimates the chunk’s size.
• If it’s too large, it finds split points (based on the shard key values).
• These split points are then sent back to the mongos to carry out the metadata
update.

Note: Chunk splitting doesn’t move data; it just updates metadata on the
config servers. Data movement happens later during the balancing phase,
when the balancer moves chunks between shards to maintain even distribution.

15
29-11-2025

Chunk Splitting Constraints


➤ Limited Legal Split Points
• MongoDB can only split chunks at points where the shard key
value changes.
• All documents with the same shard key value must stay in the same
chunk.
• This rule ensures consistency — because the shard key determines
the chunk boundaries.

Example
• If the shard key is "age" and documents are:

{"age": 13, "username": "ian"}


{"age": 13, "username": "randolph"}
------------ // possible split
{"age": 14, "username": "randolph"}
{"age": 14, "username": "eric"}
{"age": 14, "username": "hari"}
{"age": 14, "username": "mathias"}
------------ // possible split
{"age": 15, "username": "greg"}
{"age": 15, "username": "andrew"}

MongoDB can split between age:13 and age:14, or between age:14 and age:15.

16
29-11-2025

• But if all documents have the same shard key:

{"age": 12, "username": "kevin"}


{"age": 12, "username": "spencer"}
{"age": 12, "username": "alberto"}
{"age": 12, "username": "tad"}

• MongoDB cannot split this chunk — since there’s no variation in the


shard key value.
Conclusion: A good shard key must have high cardinality — i.e., many
unique or varied values — to allow effective splitting.

Config Server Availability and Split Storms


➤ Config Servers’ Role
• Chunk splits are metadata operations handled by config servers.
• When a split is needed, mongos must update the chunk metadata
on all config servers.

➤ Failure Scenario
• If one or more config servers are down, mongos:
• Cannot complete the split.
• Keeps retrying each time new writes arrive.
• This continuous retrying is called a split storm.

17
29-11-2025

Split Storm
• Cause: Config servers unavailable during chunk split attempts.
• Effect: Repeated failed attempts slow down both the mongos and the
affected shard.
• Prevention:
• Keep all config servers up and healthy.
• If necessary, restart mongos to reset its split counter (to temporarily stop
repeated attempts).

Split Tracking and Mongos Restart Issues

➤ No Global Counter
• MongoDB does not maintain a global chunk size tracker.
• Each mongos process keeps its own local counter of how much data it has written to
each chunk.

➤ Restart Effects
• When a mongos restarts:
• Its counters reset.
• If your deployment frequently spins up and down mongos instances (common in cloud setups),
chunks may never reach the split threshold.

Result:
• Chunks grow larger and larger over time (see Figures 14-5 and 14-6).

18
29-11-2025

4. Preventing Oversized Chunks


You can prevent unbounded chunk growth by:

Method Description
Avoid frequent restarts or dynamic scaling of
Keep mongos stable mongos. Keep them running continuously if
possible.
Configure a smaller chunk size than desired.
Reduce chunk size
Splits will occur earlier (at a lower threshold).
You can disable automatic chunk splitting
entirely by starting mongos with the --nosplit
Use --nosplit option
flag. Useful for testing or very controlled
environments.

19
29-11-2025

Choosing a Shard Key


CHAPTER 15
Module-4

• The most important and difficult task when using sharding is choosing
how your data will be distributed.

• Taking Stock of Your Usage


• When you shard a MongoDB collection, you choose a shard key . A
field (or combination of fields) that MongoDB uses to decide how to
divide and store your data across shards.
• Once you’ve chosen a shard key, it’s very hard to change, so you must
pick it carefully.
• To choose a good shard key, you need to think about how your
application uses data . that is, how often it reads, writes, and
queries specific fields.

1
29-11-2025

• Before picking a shard key, ask these questions:


• How many shards will your cluster have?
• With only a few shards, you can afford to send queries to all shards.
• But with many shards (e.g., hundreds), you must design queries to target specific
shards — meaning they must include the shard key in the query.

Example 1: Query with shard key - goes to one shard


• Suppose your collection orders is sharded by { userId: 1 }.
You have 10 shards.
• Example
[Link]({ userId: 12345, status: "shipped" })

• Since you included the shard key (userId), MongoDB knows exactly which
shard has this data.

• The query goes directly to that shard — fast and efficient.

Example 2: Query without shard key → goes to all shards


Same collection, but this time:
[Link]({ status: "pending" })
• You didn’t include the shard key.
• MongoDB doesn’t know which shard has the data, so it must ask every
shard.
• This is called a broadcast query — slower and heavier, especially if there
are many shards.

Summery:
• If you have only a few shards (like 3), sending the query to all shards is
not a big deal. But if you have hundreds of shards, it wastes time and
resources.
• Always include the shard key in your queries whenever possible — it
helps MongoDB send your request to the right shard quickly.

2
29-11-2025

2. What’s your goal for sharding?

• Reduce latency:
You want operations (reads/writes) to be faster — maybe by storing
data on servers closer to users or on faster machines.

• Increase throughput:
You want the system to handle more operations at the same time
(e.g., more reads/writes per second) by distributing work evenly
among shards.

• Increase total system resources:


You want more memory (RAM) or disk space available by spreading
data across multiple machines.

[Link] your shard key help achieve your goal?


Check whether it:
• Makes queries more targeted (queries can find data on specific shards
instead of searching all).
• Helps balance throughput and latency (no shard should get overloaded).
• Keeps a compact working set (frequently used data fits in memory).

Picking a shard key is like deciding how to divide work among many
workers. if you divide it wisely, everything runs fast and smooth; if you
choose poorly, some workers (shards) will be overloaded while others
sit idle.

3
29-11-2025

• Picturing Distributions
• There are three basic distributions that are the most common ways
people choose to split their data:
• Ascending key,
• Random,
• Location-based.

There are other types of keys that could be used, but most use cases fall
into one of these categories.

• Ascending shard key


• An ascending shard key is a shard key built on a field whose value
keeps increasing over time (or always moving in one direction).
Examples:
• A timestamp (e.g., creation date)
• An auto-incrementing ID (or implicitly via ObjectIds)
• Some counter/sequence field
• If you shard a collection on such a field (say { _id: 1 }, where _id is an
ObjectId or some increasing ID), then in a range-sharded cluster the
ranges (chunks) will be created in order of that field’s values.

4
29-11-2025

• Suppose we create a new document. The answer is the chunk with the
range ObjectId("5112fae0b4a4b396ff9d0ee5") through $maxKey.
• This is called the max chunk, as it is the chunk containing
$maxKey.

5
29-11-2025

6
29-11-2025

Randomly distributed shard key


• If your shard key is something like a username, email address, UUID,
or you use a “hashed” variant of a field, then the values don’t follow a
nice increasing range (like a date) or a neat geographical split.
• Instead they appear randomly and when you shard on that key, the
chunks tend to scatter more evenly across the shards.

• A shard key is a random number between 0 and 1. We’ll end up with a


random distribution of chunks on the various shards, as shown in
Figure 15-4.

7
29-11-2025

• Location-Based Shard Keys


• A location-based shard key means you choose a field in your documents
that relates to “location” in some way (geographic or otherwise).
• For example: IP address, latitude/longitude, country code, address block, or a
tenant‐region field.
• The idea: documents with similar “location” values fall into adjacent ranges
of the shard key, so they end up on the same shard or a small subset of
shards.
• This gives you data locality (i.e., related data together) and possibly better
performance for queries/operations that focus on a location.
• For instance, docs say: “Documents with shard key values close to one
another are likely to be co-located on the same shard.”
• It’s not necessarily geographic — “location” might mean logical grouping
(e.g., data region, user region, tenant zone) rather than latitude/longitude
only.

Example of How It Works


• Suppose you have a collection messages of chat messages with a
country field (say “US”, “UK”, “DE”, etc.). You pick a shard key like {
country: 1, userId: 1 }.
• The country field is your location component.
• Documents with country = “UK” will fall into chunks whose shard key
range starts with “UK”.
• You can then associate a zone (a grouping) for “UK” that maps to a
shard located in Europe, meaning all UK messages will go to that
shard.
[Link]({ country: "UK", userId: 123 }) is targeted

8
29-11-2025

Shard Key Strategies


• Hashed shard key in sharding
• For loading data as fast as possible, hashed shard keys are the best
option. A hashed shard key can make any field randomly distributed.

• You pick a field in your data to act as the “shard key” (eg. user ID).
• Instead of using the raw value to decide which shard it goes to, you
apply a hash function on that value.
• The resulting hash value is used to map the data to one of the shards.

9
29-11-2025

• A Hashed Shard Key is a type of shard key in MongoDB used to evenly


distribute data across all shards (servers).
Instead of storing data based on the raw value of the shard key,
MongoDB:
• Takes the value of the shard key field,
• Applies a hash function to that value,
• Uses the hash result to decide which shard stores that document.
• This prevents data from clustering in one shard (called a “hot shard”)
and balances the load automatically.

• Example: You have a collection called users with documents like this:

{ "_id": 1, "username": "alice", "age": 25 }


{ "_id": 2, "username": "bob", "age": 30 }
{ "_id": 3, "username": "charlie", "age": 35 }
{ "_id": 4, "username": "david", "age": 40 }

Steps:
1. Choose Shard Key
• Let’s say we want to shard based on username.
• Instead of using the plain value we’ll use a hashed shard key.

10
29-11-2025

2. Create Sharded Collection


[Link]("[Link]", { "username": "hashed" })
• This tells MongoDB to use a hash of username to decide which shard each
document goes to.

Step 3. How Data Is Distributed


username Hash(username) Assigned Shard

alice h1 (e.g., 12345) Shard A

bob h2 (e.g., -90210) Shard B

charlie h3 (e.g., 45823) Shard C

david h4 (e.g., -20202) Shard A

Even though usernames are in alphabetical order,


their hashed values are random, so data spreads evenly across shards.

Hashed shard keys are great


when:
1. You want to balance write
load evenly.
2. The data is not naturally
ordered (like usernames, user
IDs, etc.).

11
29-11-2025

• What is GridFS?

• GridFS (Grid File System) is a special specification in MongoDB for storing


and retrieving large files (like images, videos, PDFs) that exceed the 16 MB
document size limit.

• Instead of storing one big file in a single document, GridFS splits the file into
smaller chunks (default 255 KB each) and stores them across two
collections:
• [Link] → file metadata
• [Link] → actual binary chunks of data

GridFS is often used by:


• Web apps storing user-uploaded images/videos
• Backup systems
• Content management systems (CMS)

[Link] Shard Keys for GridFS


• When you store large files using GridFS, MongoDB actually creates two
collections automatically:
• [Link] — stores metadata
• [Link] — stores file data chunks
• To distribute these chunks across shards, MongoDB allows you to shard the
[Link] collection — and this is where hashed shard keys become very
useful.

• Each chunk document in [Link] looks like this:


{
"_id": ObjectId("abcd1234"),
"files_id": ObjectId("file1"), // Reference to [Link]._id
"n": 0, // Chunk number
"data": <binary chunk>
}

12
29-11-2025

Choosing the Shard Key


• For GridFS, the recommended shard key is:
{ files_id: "hashed" }
• This means MongoDB will hash the file ID to decide which shard a
chunk goes to.

Step 1: Enable sharding for your database


[Link]("mydb")

Step 2: Shard the [Link] collection


[Link]("[Link]", { "files_id": "hashed" })
• This ensures the chunks of different files are spread evenly across
shards.

Example: Let’s say you upload two large files:


• [Link] → _id: file1
• [Link] → _id: file2
• Each file is split into chunks:

files_id n data Hash(files_id) Shard

file1 0 chunk0 h1 Shard A

file1 1 chunk1 h1 Shard A

file2 0 chunk0 h2 Shard B

file2 1 chunk1 h2 Shard B

13
29-11-2025

Why Use Hashed Shard Key with GridFS?


• Balanced distribution of files
• Avoids hot shards when many files are uploaded
• Keeps chunks of each file together (important for retrieval speed)

• Never shard [Link] — it’s small and acts as metadata.


• You only shard [Link] using { files_id: "hashed" }.

A hot shard happens when one shard in a


sharded MongoDB cluster gets much more
traffic (reads or writes) than the others.
This makes that shard “hot” . overworked , while
others stay underused.

3. Firehose Strategy?
• The Firehose Strategy refers to a data collection approach where all
raw data is captured continuously and stored, even before deciding
what to analyze.
• Instead of selectively collecting only certain data, you collect it all
first, and decide later what’s useful.

Steps:
• you don’t pre-filter what you collect.
• You store all raw logs, sensor readings, user events, clickstreams, etc.
• Later, you process and analyze it for insights.

14
29-11-2025

Example
• Suppose you run an e-commerce website.
• Instead of logging only:
{ "user_id": 123, "item_id": 10, "action": "purchase" }

You collect everything users do:


{ "user_id": 123, "item_id": 10, "action": "view", "timestamp": "10:00:01" }
{ "user_id": 123, "item_id": 10, "action": "add_to_cart", "timestamp": "10:00:05" }
{ "user_id": 123, "item_id": 10, "action": "purchase", "timestamp": "10:00:10" }

You store all of this in a data lake or a stream (like Kafka ).


Later, your analytics team can decide:
• To train models on purchase prediction
• To compute conversion rates

Concept Firehose Strategy

Data approach Capture everything, filter later

Used for Logs, streams, sensors, click data

Benefit No data loss, flexible analysis

Challenge High volume, storage cost

Summery:

15
29-11-2025

4. Multi-Hotspot:
• A multi-hotspot occurs when multiple keys or ranges in your sharded or
distributed database receive not equally distributed high traffic at the
same time.
• Unlike a single hot shard (where one shard is overloaded), a multi-
hotspot means several shards or key ranges are all “hot” each getting
heavy, uneven load.

Example:
• Imagine a highway with many toll booths (shards).
• If one booth has a huge line - that’s a hot spot.
• If many booths have big lines at once - that’s a multi-hotspot.

Example
Suppose you have a social media app storing posts in MongoDB:
{ "user_id": 101, "post": "..." }
• You shard using user_id as the key:
[Link]("[Link]", { "user_id": 1 })
• Now imagine:
• Users 101, 102, and 103 are celebrities with millions of followers.
• Each gets massive read/write traffic.

• Then:
• Shards holding user_id 101, 102, and 103 become hot simultaneously.
• That’s a multi-hotspot — multiple busy shards, while others are mostly
idle.

16
29-11-2025

Concept Description

Hot Shard One shard overloaded

Multi-Hotspot Multiple shards overloaded

Main Cause Uneven data or traffic distribution

Hashed / compound shard keys, caching, rate


Best Fix
limiting

• A standalone MongoDB server works fastest when you insert data in


ascending order (for example, increasing _id values).
• But in a sharded cluster, writes need to be spread across shards to avoid
overloading any one shard.

This creates a conflict:


• Standalone = best with ascending writes
• Sharded = best with distributed writes
To fix this, MongoDB uses a compound shard key that helps balance both.

Note:
• The first part of the shard key is a random value (to spread writes across
shards).
• The second part is a sequential or ascending value (to keep inserts
efficient within each shard).

17
29-11-2025

Example
• Suppose your shard key is:
{ "region": 1, "timestamp": 1 }
• region = random or low-cardinality value (e.g., North, South, East,
West)
• timestamp = ascending value (insert order)

Then:
• Different regions spread writes across shards
• Timestamps within each region stay in order

• Result:
Even load across the cluster and efficient inserts inside each shard.

• Shard Key Rules and Guidelines


 When choosing a shard key, you must follow certain rules and guidelines similar to
selecting an index.
 A shard key determines how data is distributed across shards, so it should be chosen
carefully.
 Shard keys cannot be arrays, and once a document is inserted, its shard key value
cannot be changed without deleting and reinserting the document.
Note: Therefore, select a field that rarely changes. Also, special index types like
geospatial indexes cannot be used, though hashed indexes are allowed.

18
29-11-2025

• Shard Key Rules and Guidelines


 The shard key should have high cardinality, meaning it should contain
many unique values to distribute data evenly(A high-cardinality shard
key means the key has many unique values, allowing MongoDB to evenly
distribute data across all shards).

 If a field has few unique values (like a "logLevel" field with only "DEBUG",
"WARN", or "ERROR"), MongoDB cannot create many chunks.

 In such cases, you can use a compound shard key combining a low-
cardinality field with a high-cardinality one (e.g., "logLevel" and
"timestamp") to ensure better data distribution.

• Controlling Data Distribution


• In a sharded MongoDB cluster, data is automatically distributed
across shards based on the shard key.
Normally, MongoDB decides where data goes and keeps the load
balanced automatically.
• But sometimes, this automatic distribution doesn’t fit your specific
needs especially when:
• You want some data to live on specific servers (e.g., expensive vs. cheap
hardware).
• You want certain collections to be isolated for performance or security
reasons.
• You’re managing a small cluster and want fine-grained control.
• In these cases, MongoDB allows you to manually influence or control
data placement.
• As your cluster grows, manual control becomes harder — because
balancing thousands of chunks manually is inefficient.

19
29-11-2025

• Cluster for Multiple Databases and Collections


Default Behavior
• By default, MongoDB distributes all collections evenly across all
shards in a cluster.
• This works fine if all data types are similar in value and usage
(homogeneous data)

You May Want Manual Control


• Some data (like logs) may be low priority — you don’t want it
stored on costly, high-performance shards.
• Some shards might be powerful — you want them reserved for
important collections only.

Shard Tagging
• You can tag shards and tell
Shard ID Tag Meaning MongoDB to store specific
High-performance collections on shards with certain
shard0000 high
shard tags.
shard0001–
(none) Default shards
0003
shard0004, [Link]("shard0000", "high")
low Low-cost shards
shard0005 [Link]("shard0004", "low")
[Link]("shard0005", "low")

20
29-11-2025

Assign Collections to Tagged Shards


(a) High-Priority Collection
[Link]("[Link]",
{"shardKey": MinKey},
{"shardKey": MaxKey},
"high")
• All data in [Link] collection goes only to the shard(s)
tagged "high".

Balancer Migration
• Tag assignment does not move data instantly.
• The balancer gradually moves data to match the new tag
configuration.
• So, existing chunks may take some time to migrate.

Creating a New Tag Group


• If you want another group (for example, shards not tagged "high"):
[Link]("shard0001", "whatever")
[Link]("shard0002", "whatever")
[Link]("shard0003", "whatever")
[Link]("shard0004", "whatever")
[Link]("shard0005", "whatever")

21
29-11-2025

• Then assign the collection:


[Link]("[Link]",
{"shardKey": MinKey},
{"shardKey": MaxKey},
"whatever")
• Now [Link] is distributed across these five shards (not on the
“high” shard).

Tagging allows you to tell MongoDB which shards should store which collections,
giving you more control over how data is distributed across your cluster.

Action Command Purpose

Add tag to a shard [Link]("shard0000", "high") Label shards based on use

Assign collection to a tag [Link]("coll", {...}, {...}, "tag") Control where data is stored

Remove a tag [Link]("shard0000", "high") Undo a tag assignment

Edit manually Use [Link] and [Link] Manage tag data directly

22

You might also like