Sharding

A replica set keeps your data alive, but every member still holds the entire dataset. That works right up until one machine can no longer fit all your data, or one primary can no longer handle all your writes. Adding memory and faster disks — scaling up — has a ceiling. Sharding scales out: it splits one logical collection across many machines, so the data and the write load divide among them.

Everything below ran on this Mac on a throwaway cluster of four mongo:8.3.11 containers on a Docker network, each capped at 512 MB and removed afterwards: one config server (a one-node replica set), two shards (shardA, shardB, one node each) and one mongos, with mongosh 2.11.1. It is a teaching cluster — production shards and config servers are three-node replica sets — and the course’s automated checker does not run these fences (they need the cluster), so their outputs are recorded from the real run. Python/Go/Rust tabs mirror the shell call and are not executed by the checker.

The lab

scripts/shard-lab.sh

docker network create mongodb-shard-lab
run() { docker run -d --name "$1" --hostname "$1" --network mongodb-shard-lab --memory 512m mongo:8 "${@:2}"; }
run cfg     mongod --configsvr --replSet cfg    --bind_ip_all --wiredTigerCacheSizeGB 0.25
run shard-a mongod --shardsvr  --replSet shardA --bind_ip_all --wiredTigerCacheSizeGB 0.25
run shard-b mongod --shardsvr  --replSet shardB --bind_ip_all --wiredTigerCacheSizeGB 0.25
docker exec cfg     mongosh --quiet --port 27019 --eval 'rs.initiate({ _id: "cfg", configsvr: true, members: [{ _id: 0, host: "cfg:27019" }] })'
docker exec shard-a mongosh --quiet --port 27018 --eval 'rs.initiate({ _id: "shardA", members: [{ _id: 0, host: "shard-a:27018" }] })'
docker exec shard-b mongosh --quiet --port 27018 --eval 'rs.initiate({ _id: "shardB", members: [{ _id: 0, host: "shard-b:27018" }] })'
run mongos  mongos --configdb cfg/cfg:27019 --bind_ip_all
docker exec mongos mongosh --quiet --eval 'sh.addShard("shardA/shard-a:27018"); sh.addShard("shardB/shard-b:27018")'
# clean up: docker rm -f cfg shard-a shard-b mongos && docker network rm mongodb-shard-lab

The pieces of a sharded cluster

A shard is a replica set holding one portion of the data. The config servers are their own replica set that stores the cluster metadata — which chunk lives on which shard. mongos is the router your application connects to instead of any single shard; it reads that metadata and forwards each request to the right shard or shards.

flowchart TB
  App["Application"] --> M["mongos router"]
  M --> CFG["Config servers — store the chunk map"]
  M --> Sh1["Shard 1 (replica set)"]
  M --> Sh2["Shard 2 (replica set)"]
  M --> Sh3["Shard 3 (replica set)"]
  CFG -. "tells mongos where each chunk lives" .- M

The application talks only to mongos, which uses the config servers' map to route requests to the right shard

The cluster as mongos sees it:

// mongosh/sh-topology.js @skip-verify needs the sharded lab cluster (real output below)
console.log(db.getSiblingDB("config").shards.find({}, { host: 1 }).toArray());
console.log("balancer enabled:", sh.getBalancerState(), "| mode:", db.adminCommand({ balancerStatus: 1 }).mode);

[
  { _id: 'shardA', host: 'shardA/shard-a:27018' },
  { _id: 'shardB', host: 'shardB/shard-b:27018' }
]
balancer enabled: true | mode: full

Sharding a collection — enableSharding is optional

Older tutorials (and older versions of this lesson) start with sh.enableSharding("library"). The manual: “Starting in MongoDB 6.0, this command is not required to shard a collection. The enableSharding command explicitly creates a database” (enableSharding, consulted 2026-09-30). Proof on 8.3.11 — the very first command on the fresh cluster:

mongosh

// mongosh/sh-shard-collection.js @skip-verify needs the sharded lab cluster (real output below)
const r = sh.shardCollection("library.loans", { memberId: 1 });   // no enableSharding first
console.log("shardCollection ok:", r.ok);
console.log(db.getSiblingDB("config").databases.find({ _id: "library" }, { primary: 1 }).toArray());

Node.js

// app/sh-shard-collection.mjs @skip-verify needs the sharded lab cluster
const admin = client.db("admin");
await admin.command({ shardCollection: "library.loans", key: { memberId: 1 } });

Python

admin = client.admin
admin.command(
    "shardCollection", "library.loans", key={"memberId": 1}
)

Go

admin := client.Database("admin")
if err := admin.RunCommand(ctx, bson.D{
    {Key: "shardCollection", Value: "library.loans"},
    {Key: "key", Value: bson.D{{Key: "memberId", Value: 1}}},
}).Err(); err != nil {
    return err
}

Rust

let admin = client.database("admin");
admin
    .run_command(doc! {
        "shardCollection": "library.loans",
        "key": { "memberId": 1 },
    })
    .await?;

shardCollection ok: 1
[ { _id: 'library', primary: 'shardB' } ]

The database was created on demand and given a primary shard (shardB in this run; the cluster picks it) — the home of its unsharded collections. That is all enableSharding still does.

The shard key, chunks, and the balancer

The shard key is a field (or fields) MongoDB uses to decide which shard a document belongs to. It divides the key’s value range into contiguous chunks and assigns each chunk to a shard; a background balancer migrates chunks to keep shards even. Ranged sharding keeps nearby values together (good for range queries, risky if writes cluster at one end); hashed sharding hashes the key first (even writes, but ranges fan out to every shard).

flowchart LR
  K["Shard key value"] --> C["Which chunk owns this value"]
  C --> S["Which shard owns that chunk"]
  S --> Done["Document lands on that shard"]
  Bal["Balancer"] -. "migrates chunks to keep shards even" .- S

A document's shard key decides its chunk, and the chunk decides its shard; the balancer evens out chunk counts over time

Targeted versus scatter-gather, in explain

A query that carries the shard key is targeted; one without it is scatter-gather. Below, 100 000 documents loaded into a hashed collection (events, key deviceId), then explain on mongos:

// mongosh/sh-route.js @skip-verify needs the sharded lab cluster (real output below)
const lib = db.getSiblingDB("library");
lib.events.drop();
sh.shardCollection("library.events", { deviceId: "hashed" });
let batch = [];
for (let i = 0; i < 100000; i++) {
  batch.push({ deviceId: "dev-" + (i % 5000), i });
  if (batch.length === 10000) { lib.events.insertMany(batch, { ordered: false }); batch = []; }
}
function route(label, cursor) {
  const p = cursor.explain().queryPlanner.winningPlan;
  console.log(label, "->", p.stage, (p.shards ?? []).map((s) => s.shardName));
}
route("deviceId = dev-42 (shard key, equality)", lib.events.find({ deviceId: "dev-42" }));
route("i = 42            (no shard key)       ", lib.events.find({ i: 42 }));
route("deviceId range    (hashed key, range)  ", lib.events.find({ deviceId: { $gte: "dev-1", $lt: "dev-2" } }));

deviceId = dev-42 (shard key, equality) -> SINGLE_SHARD [ 'shardA' ]
i = 42            (no shard key)        -> SHARD_MERGE [ 'shardA', 'shardB' ]
deviceId range    (hashed key, range)   -> SHARD_MERGE [ 'shardA', 'shardB' ]

Equality on the shard key went to exactly one shard (SINGLE_SHARD); no shard key, or a range over a hashed key, needed both (SHARD_MERGE). That is the price of hashing: order is destroyed.

Choosing a good shard key

A bad shard key cannot be quietly fixed later (see resharding below). Three properties matter: high cardinality (many distinct values, so many chunks), even distribution (no value dominates), and no hotspot on writes. A steadily increasing key — a timestamp, an auto-increment id — is the classic trap.

Measuring a candidate: analyzeShardKey

Since MongoDB 7.0, analyzeShardKey “calculates metrics for evaluating a shard key for an unsharded or sharded collection” (analyzeShardKey, consulted 2026-09-30) — this is the tool the older lesson called a “Shard Key Advisor”. Its key characteristics run on the data already there (read/write distribution metrics additionally need query sampling via configureQueryAnalyzer). Two candidates over the same 20 000 documents:

// mongosh/sh-analyze.js @skip-verify needs the sharded lab cluster (real output below)
const lib = db.getSiblingDB("library");
lib.readings.drop();
lib.readings.insertMany(Array.from({ length: 20000 }, (_, i) => ({ ts: i, deviceId: "dev-" + (i % 5000) })));   // unsharded copy to analyse
lib.readings.createIndex({ ts: 1 });
lib.readings.createIndex({ deviceId: 1 });
function analyze(key) {
  const k = db.adminCommand({ analyzeShardKey: "library.readings", key }).keyCharacteristics;
  return { key, numDocsTotal: Number(k.numDocsTotal), numDistinctValues: Number(k.numDistinctValues), mostCommonFrequency: Number(k.mostCommonValues[0].frequency), monotonicity: k.monotonicity.type };
}
console.log(analyze({ ts: 1 }));
console.log(analyze({ deviceId: 1 }));

{
  key: { ts: 1 },
  numDocsTotal: 20000,
  numDistinctValues: 20000,
  mostCommonFrequency: 1,
  monotonicity: 'monotonic'
}
{
  key: { deviceId: 1 },
  numDocsTotal: 20000,
  numDistinctValues: 5000,
  mostCommonFrequency: 4,
  monotonicity: 'not monotonic'
}

ts is flagged monotonic: every new document sorts after all the others, so all inserts land on the top chunk. deviceId is not monotonic, has 5 000 distinct values and its most common value occurs 4 times — a reasonable key. (The tool describes the data; it does not decide for you.)

Changing your mind later

The old advice — “changing the shard key requires rebuilding the collection” — is stale. reshardCollection (MongoDB 5.0) changes a collection’s key online, and 8.0 adds moveCollection (move an unsharded collection to another shard) and unshardCollection (resharding, moveCollection, unshardCollection, consulted 2026-09-30). It still costs: the manual states “The minimum duration of a resharding operation is always 5 minutes”, only one collection can be resharded at a time, and writes get a brief critical section. Real run (a 10 000-document collection):

// mongosh/sh-reshard.js @skip-verify needs the sharded lab cluster and ~5 minutes (real output below)
const adm = db.getSiblingDB("admin");
const lib = db.getSiblingDB("library");
lib.tickers.drop();
lib.tickers.insertMany(Array.from({ length: 10000 }, (_, i) => ({ ts: i, deviceId: "dev-" + (i % 5000) })));
lib.tickers.createIndex({ ts: 1 });                       // a populated collection needs an index on the key first
sh.shardCollection("library.tickers", { ts: 1 });
let t = Date.now();
console.log("reshardCollection ts:1 -> deviceId:hashed ok:", adm.runCommand({ reshardCollection: "library.tickers", key: { deviceId: "hashed" } }).ok, Math.round((Date.now() - t) / 1000) + " s");
console.log(db.getSiblingDB("config").collections.find({ _id: "library.tickers" }, { key: 1 }).toArray());

reshardCollection ts:1 -> deviceId:hashed ok: 1 331 s
[ { _id: 'library.tickers', key: { deviceId: 'hashed' } } ]

In Compass / Atlas: Compass shows a sharded collection’s shard key on the Indexes/Schema view; Atlas exposes sharding as a cluster option. Both drive the same commands — analyzeShardKey and reshardCollection — shown above.

Tips and gotchas

• Get the key right early, but know it is changeable: resharding works, at least five minutes and with a cost.
• Do not shard too early. One replica set serves a great deal of traffic; shard on evidence.
• Carry the shard key in common queries — measured above: targeted SINGLE_SHARD versus SHARD_MERGE on both shards.
• Sharding does not fix a query without an index; it multiplies the scan by the number of shards.

🍃 The shard key is the decision that matters

Measure candidates with analyzeShardKey, distribute writes with a high-cardinality key, and avoid monotonic keys unless hashed. The measurement below is the argument: a timestamp key put 100 000 of 100 000 documents on one shard.

Reading AI-generated code

An assistant asked to “shard the readings collection for time-series queries” picks the timestamp:

🐛 Spot the bug

// AI comment: "ts has 100000 distinct values (high cardinality) and supports time-range queries, so it spreads data evenly"
sh.shardCollection("library.readings", { ts: 1 });

Show the answer

High cardinality is necessary, not sufficient. ts only ever increases, so every insert falls in the highest range, which lives on one shard. Real run — 100 000 documents inserted through mongos with each key:

// mongosh/sh-hotspot.js @skip-verify needs the sharded lab cluster (real output below)
const lib = db.getSiblingDB("library");
for (const c of ["readings_ranged", "readings_hashed"]) lib[c].drop();
sh.shardCollection("library.readings_ranged", { ts: 1 });               // the AI's choice
sh.shardCollection("library.readings_hashed", { deviceId: "hashed" });  // the fix
let a = [], b = [];
for (let i = 0; i < 100000; i++) {
  a.push({ ts: i, deviceId: "dev-" + (i % 5000) });
  b.push({ ts: i, deviceId: "dev-" + (i % 5000) });
  if (a.length === 10000) { lib.readings_ranged.insertMany(a, { ordered: false }); lib.readings_hashed.insertMany(b, { ordered: false }); a = []; b = []; }
}
for (const c of ["readings_ranged", "readings_hashed"]) {
  console.log(c, lib[c].aggregate([{ $collStats: { storageStats: {} } }, { $project: { _id: 0, shard: 1, docs: "$storageStats.count" } }]).toArray());
}

readings_ranged [ { shard: 'shardB', docs: 100000 } ]
readings_hashed [ { shard: 'shardB', docs: 51100 }, { shard: 'shardA', docs: 48900 } ]

The ranged timestamp key sent everything to one shard (the primary shard, shardB here); the hashed key split it about half and half. The fix has its own cost (measured earlier): a time-range query over a hashed key becomes scatter-gather. If you need both, a compound key such as { deviceId: 1, ts: 1 } or a hashed prefix is the usual answer — analyze it before you commit.

Mental Model

A sharded collection is a sorted list of key ranges (chunks), each range owned by one shard, kept in the config servers; mongos is a lookup table in front of it. Everything follows from how your key maps onto that list: equality lands on one range (targeted), a range of the key lands on a few ranges if the key is ordered (ranged) or on all of them if it is hashed, and a value that always increases lands on the last range forever (hotspot).

Trade-offs

Choice	Benefit	Cost
Ranged sharding	Nearby key values live in the same chunks, so a range over the key needs few of them	A monotonic key hotspots: 100 000 of 100 000 documents on one shard
Hashed sharding	Even writes (48 900 / 51 100 in the run)	Range queries and non-key queries become SHARD_MERGE on every shard
Sharding early	Room to grow	Operational complexity (four processes here, more in production); every query needs the key or scatters
Resharding later	Key can be changed online since 5.0	At least 5 minutes per operation, one collection at a time

Common Misconceptions

• “You must run sh.enableSharding() first.” — not since 6.0; the first shardCollection created the database itself.
• “High cardinality is enough for a shard key.” — ts had 20 000 distinct values in the analysis (100 000 in the load) and still put every document on one shard.
• “The shard key cannot be changed.” — reshardCollection changes it online (5.0+), taking at least five minutes.
• “Adding shards speeds up any query.” — a query without the shard key visits every shard; and a query without an index scans on each.
• “There is a Shard Key Advisor command.” — the server tool is analyzeShardKey (7.0+); anything called an “advisor” is a UI on top of it.
• “mongos stores data.” — it is a stateless router; data is on the shards, metadata on the config servers.

Python/Go/Rust tabs mirror the shell call; they are not executed by the course’s automated checker.

What does the application connect to in a sharded cluster?

A randomly chosen shard The mongos router The config servers directly The balancer

On MongoDB 8.3 you run sh.shardCollection("library.loans", { memberId: 1 }) on a fresh cluster with no enableSharding. What happens?

It fails: the database is not sharding-enabled It succeeds: the database is created on demand with a primary shard It asks for confirmation It shards every collection in the database

You shard on { ts: 1 } where ts is a steadily increasing number and insert 100 000 documents. Where do they land?

Evenly on both shards All on one shard (the one owning the top range) On the config servers They are rejected

Which query is targeted at one shard on a collection hashed on deviceId?

find({ i: 42 }) find({ deviceId: "dev-42" }) find({ deviceId: { $gte: "dev-1", $lt: "dev-2" } }) None: hashed collections always scatter

Is it true that changing a shard key requires rebuilding the collection?

Yes, always No: reshardCollection (5.0+) changes it online, though an operation takes at least 5 minutes No: it changes instantly with sh.setKey Only for hashed keys

Submit

Mark as complete