--- title: Sharding source: 'https://academia.sh/en/courses/nosql/sharding' course: 'Non-Relational Data Models' language: en updated: '2026-08-23T07:00:46+00:00' license: 'CC BY-SA 4.0' --- # Sharding Choosing the shard key from a document field: translating the key's presence-in-every-document and uniqueness requirements into an unroutable document count and an upper bound on shard count, measuring five candidate keys on the same twenty-thousand-document loan collection for placement balance and hot shards, counting how many shards the same query mix touches under each candidate, and showing how a compound key without a matching prefix scatters a query. The previous lesson showed what adding a member to the cluster solves and what it does not: when a member is lost, service continues, but every member keeps carrying the entirety of the same data. Once the catalog grows and loan records no longer fit on a single member's disk, replication has nothing left to offer; the data has to be split across members. The name for this split is sharding, and it comes down to a single decision: choosing the field that determines which shard each document goes to. That field is called the **shard key**. The arithmetic of data-splitting strategies — whether the split is range-based or hash-based, how much data a rebalance moves — was measured in the Scaling the Data Layer course and is not repeated here. The question here is specific to the document model: the key is a document field, and choosing that field is an expensive decision to reverse. ## The Key Comes From Inside the Document In a relational table, the partitioning column is part of the schema and is present in every row. The document model offers no such guarantee: documents in the same collection can carry different sets of fields. This places two conditions on the shard key. **The key field must be present in every document.** The router cannot send a document that lacks the field to any shard at all. In the library database this is not an abstract risk: some of the loan records migrated from the legacy system have no branch field at all. **The key's uniqueness bounds the shard count.** The smallest unit of sharding is a single key value; documents that share a value cannot be split apart. A field with six distinct values cannot be spread across more than six shards, and if one of those values carries half the documents, that pile never gets split. This is an upper bound that is independent of the run. The third condition is the one this lesson measures: the key has to spread writes across shards while still letting frequent queries route to a single shard. If a query pins the key's prefix, the router sends it to one shard; this is called a **targeted query**. If it does not, the router has to ask every shard; this is called a **scatter-gather query**. The difference between the two kinds is the number of shards touched, and it can be counted. ## The Mechanism **NS16 — the distribution of loan records across branches is 45, 18, 14, 11, 8, and 4 units.** Reason: the central branch carries the largest collection and the most members. **NS17 — 2 percent of the migrated legacy records have no branch field.** **NS18 — query mix: the member query is 45 percent, the branch end-of-day listing 20 percent, the last seven days' loans 15 percent, access by record id 15 percent, the overdue-loan scan 5 percent.** All three assumptions have a linear effect; if the weights in the mix change, the weighted average below shifts in the same direction. ```js // shard/model.mjs — loan document generator and shard placement. The generator is // handwritten and the seed is visible: every run produces the same document set. The // shard count P is a parameter; range- vs. hash-based placement was measured in the // Scaling the Data Layer course. Fixed here: every document routes by its KEY FIELD VALUE. export const BRANCH = [["central", 45], ["kadikoy", 18], ["besiktas", 14], ["uskudar", 11], ["sisli", 8], ["adalar", 4]]; export const DAY = 180, MEMBER = 6000; export function generate({ count = 20000, seed = 20260731, missingBranch = 0.02 } = {}) { let s = seed % 2147483647; const rand = () => (s = (s * 48271) % 2147483647) / 2147483647; const total = BRANCH.reduce((t, [, w]) => t + w, 0); const docs = []; for (let i = 1; i <= count; i += 1) { let x = rand() * total, branch = BRANCH[0][0]; for (const [name, w] of BRANCH) { if (x < w) { branch = name; break; } x -= w; } const d = { loan_id: i, member_id: 1 + Math.floor(rand() * MEMBER), borrow_day: 1 + Math.floor(rand() * DAY), returned: rand() < 0.82 }; if (rand() >= missingBranch) d.branch = branch; // some legacy records have no branch field docs.push(d); } return docs; } const compare = (a, b) => { // total ordering over key tuples for (let i = 0; i < a.length; i += 1) if (a[i] !== b[i]) return a[i] < b[i] ? -1 : 1; return 0; }; const hashValue = (m) => { let h = 2166136261; for (const c of m) h = Math.imul(h ^ c.charCodeAt(0), 16777619); return h >>> 0; }; // Shard placement: key values are sorted and split into P windows with an equal-doc-count target. // A single key value cannot be split — the smallest unit of sharding is one key value. export function place(docs, fields, P, hashed = false) { const key = (d) => fields.map((f) => d[f]); const complete = docs.filter((d) => fields.every((f) => d[f] !== undefined)); const counts = new Map(); for (const d of complete) { const k = JSON.stringify(hashed ? [hashValue(JSON.stringify(key(d)))] : key(d)); counts.set(k, (counts.get(k) ?? 0) + 1); } const sorted = [...counts.entries()].sort((a, b) => compare(JSON.parse(a[0]), JSON.parse(b[0]))); const target = complete.length / P; const placement = new Map(); let p = 0, cumulative = 0; for (const [k, n] of sorted) { placement.set(k, p); cumulative += n; if (cumulative >= target * (p + 1) && p < P - 1) p += 1; } const shardOf = (d) => placement.get(JSON.stringify(hashed ? [hashValue(JSON.stringify(key(d)))] : key(d))); const distribution = Array(P).fill(0); for (const d of complete) distribution[shardOf(d)] += 1; return { shardOf, distribution, complete, missing: docs.length - complete.length, distinct: counts.size }; } // A query can be routed if it pins a PREFIX of the key (targeted query); if it does not, the // router has to go to every shard (scatter-gather query). Under hash placement a range clause // cannot be used, because adjacent values do not land on adjacent shards. export function touched(query, fields, hashed, y, P) { const usable = []; for (const f of fields) { if (query.eq && f in query.eq) { usable.push({ field: f, type: "eq" }); continue; } if (query.range && query.range.field === f && hashed === false) usable.push({ field: f, type: "range" }); break; } if (usable.length === 0) return P; const matches = (d) => usable.every((u) => u.type === "eq" ? d[u.field] === query.eq[u.field] : d[u.field] >= query.range.min && d[u.field] <= query.range.max); return new Set(y.complete.filter(matches).map(y.shardOf)).size; } ``` ```js // shard/measure.mjs — five shard-key candidates: placement balance, hot shard, touched shards import { generate, place, touched, BRANCH, DAY, MEMBER } from "./model.mjs"; const P = 4, COUNT = 20000; const docs = generate({ count: COUNT }); const CANDIDATE = [["branch", ["branch"], false], ["borrow_day", ["borrow_day"], false], ["hash(borrow_day)", ["borrow_day"], true], ["member_id", ["member_id"], false], ["{branch,member_id}", ["branch", "member_id"], false]]; const QUERY = [ ["S1 member's open loans", 45, (i) => ({ eq: { member_id: 1 + (i * 137) % MEMBER } })], ["S2 branch end-of-day listing", 20, (i) => ({ eq: { branch: BRANCH[i % BRANCH.length][0], borrow_day: 1 + (i * 31) % DAY } })], ["S3 last 7 days' loans", 15, () => ({ range: { field: "borrow_day", min: DAY - 6, max: DAY } })], ["S4 loan record by id", 15, (i) => ({ eq: { loan_id: 1 + (i * 911) % COUNT } })], ["S5 overdue-loan scan", 5, () => ({ eq: {} })], ]; const s = (x, n) => String(x).padStart(n); const recent = docs.filter((d) => d.borrow_day > DAY - 7); console.log(`${COUNT} loan documents, ${P} shards, ${BRANCH.length} branches, ${MEMBER} members, ${DAY} days. Seed 20260731.`); console.log(`taken in the last 7 days: ${recent.length} documents\n`); console.log("shard key | distinct vals | unroutable | shard distribution | busiest/avg | last-7-day busiest shard"); console.log("--------------------|---------------|------------|-------------------------|-------------|-------------------------"); const placement = new Map(); for (const [name, fields, hashed] of CANDIDATE) { const y = place(docs, fields, P, hashed); placement.set(name, y); const fullest = Math.max(...y.distribution) / (y.complete.length / P); const rd = Array(P).fill(0); for (const d of recent) if (fields.every((f) => d[f] !== undefined)) rd[y.shardOf(d)] += 1; const share = Math.max(...rd) / rd.reduce((t, x) => t + x, 0); console.log(`${name.padEnd(19)} | ${s(y.distinct, 12)} | ${s(y.missing, 17)} | ` + `${y.distribution.map((x) => s(x, 5)).join(" ").padEnd(23)} | ${s(fullest.toFixed(2), 11)} | ` + `${s("%" + (100 * share).toFixed(1), 23)}`); } console.log("\nshards touched (per query, 1 = targeted, 4 = scatter-gather):"); console.log("shard key | S1 | S2 | S3 | S4 | S5 | weighted average"); console.log("--------------------|------|------|------|------|------|-----------------"); for (const [name, fields, hashed] of CANDIDATE) { const y = placement.get(name); const cell = [], weighted = []; for (const [, weight, example] of QUERY) { let t = 0; for (let i = 0; i < 12; i += 1) t += touched(example(i), fields, hashed, y, P); cell.push(t / 12); weighted.push((t / 12) * weight); } const avg = weighted.reduce((a, b) => a + b, 0) / QUERY.reduce((a, b) => a + b[1], 0); console.log(`${name.padEnd(19)} | ${cell.map((x) => s(x.toFixed(2), 4)).join(" | ")} | ${s(avg.toFixed(2), 16)}`); } console.log("\nquery mix: " + QUERY.map(([a, w]) => `${a.slice(0, 2)} %${w}`).join(", ")); console.log(`independent of the run: the number of distinct values of a key is the upper bound` + ` on shard count (${BRANCH.length} for branch).`); ``` ``` 20000 loan documents, 4 shards, 6 branches, 6000 members, 180 days. Seed 20260731. taken in the last 7 days: 782 documents shard key | distinct vals | unroutable | shard distribution | busiest/avg | last-7-day busiest shard --------------------|---------------|------------|-------------------------|-------------|------------------------- branch | 6 | 406 | 12328 3489 1597 2180 | 2.52 | %63.8 borrow_day | 180 | 0 | 5022 5031 5031 4916 | 1.01 | %100.0 hash(borrow_day) | 180 | 0 | 5034 4973 5002 4991 | 1.01 | %59.6 member_id | 5788 | 0 | 5000 5000 5002 4998 | 1.00 | %27.0 {branch,member_id} | 13443 | 406 | 4899 4898 4899 4898 | 1.00 | %25.8 shards touched (per query, 1 = targeted, 4 = scatter-gather): shard key | S1 | S2 | S3 | S4 | S5 | weighted average --------------------|------|------|------|------|------|----------------- branch | 4.00 | 1.00 | 4.00 | 4.00 | 4.00 | 3.40 borrow_day | 4.00 | 1.00 | 1.00 | 4.00 | 4.00 | 2.95 hash(borrow_day) | 4.00 | 1.00 | 4.00 | 4.00 | 4.00 | 3.40 member_id | 1.00 | 4.00 | 4.00 | 4.00 | 4.00 | 2.65 {branch,member_id} | 4.00 | 1.50 | 4.00 | 4.00 | 4.00 | 3.50 query mix: S1 %45, S2 %20, S3 %15, S4 %15, S5 %5 independent of the run: the number of distinct values of a key is the upper bound on shard count (6 for branch). ``` ## Placement Balance and Hot Shards **The most intuitive candidate gives the worst balance.** Splitting loan records by branch looks like it matches the operational structure; the measurement says something else. When six branches are squeezed into four shards, the shards end up with 12,328, 3,489, 1,597, and 2,180 documents: the busiest shard holds 2.52 times the average. The reason is uniqueness — the smallest grain a shard can carry is one branch, and the central branch alone produces 45 percent of all loan records. Adding more shards does not fix the imbalance either, because six distinct values are the upper bound for six shards. The row's second number is a warning of its own: 406 documents carry no branch field and cannot be routed to any shard. **A balanced placement is not enough on its own.** The `borrow_day` key splits documents almost perfectly — the busiest shard holds 1.01 times the average. The last column refutes this row: 100 percent of the 782 loans taken in the last seven days fall onto a single shard. When a monotonically increasing field is the key, past data spreads evenly, but every new write piles onto the shard carrying the most recent window. This is called a **hot shard**, and its only visible symptom in the store is that one member queues writes while the cluster's other three sit idle. **Hash placement trades the hot shard for something else.** Hashing the same field brings the hot shard's share down from 100 percent to 59.6 percent. The reason it does not drop to 25 percent is, again, uniqueness: when seven days' seven distinct values spread across four shards, some shards end up with two days. The cost sits in the second table, and it is large. **The candidate with the highest uniqueness gives the best balance.** Under the `member_id` key, the distribution is 5,000, 5,000, 5,002, 4,998, and the last-seven-days share is 27 percent — close to the 25 percent that is ideal for four shards. The compound key gives the same balance but brings back the 406 unroutable documents, because branch sits in its prefix. ## Targeted and Scatter-Gather Queries The second table runs the same query mix across all five candidates, and it reverses the decision that balance alone would suggest. The `member_id` key gives the best result on the weighted average, 2.65 shards, because the mix's heaviest query — the member query — pins the key exactly: 1.00 shard. Under the same key, the branch end-of-day listing rises to 4.00 — because it never pins the key at all, it goes to all four shards at once. The compound key `{branch,member_id}` is this lesson's most instructive row. Its placement balance is perfect (1.00) and the branch query drops to 1.50 shards; against that, the member query rises to 4.00. The reason is that a compound key works left to right: the member query pins the `member_id` field but does not pin the `branch` field in the prefix, so the router cannot tell which shard to go to. Its weighted average, 3.50, is the worst of the five candidates. Adding a compound key does not narrow access — it tightens the prefix condition. Comparing `borrow_day` against its hash placement shows the trade-off in a single column. Hash placement brought the hot shard down from 100 percent to 59.6 percent; in exchange, the last-seven-days query rose from 1.00 to 4.00 shards, and the weighted average climbed from 2.95 to 3.40. Because hash placement does not put adjacent values on adjacent shards, the range query cannot be routed. The thing that spreads writes and the thing that collects reads come into direct conflict here. The last line restates the bound that is independent of the run: the number of distinct values a key has is the upper bound on shard count. This bound does not depend on the machine, the data, or the configuration. ## Summary - The shard key is a document field, and a document that lacks the field cannot be routed: under the branch-based key, 406 of the 20,000 documents could not go to any shard. - The number of distinct values a key has is the upper bound on shard count; in the collection split across six branches, the busiest shard held 2.52 times the average, and adding more shards does not fix this. - A balanced placement does not prevent a hot shard: under the `borrow_day` key, placement was balanced to within 1.01 times even as 100 percent of the 782 loans from the last seven days fell onto a single shard. - Hash placement brought the hot shard's share down to 59.6 percent but raised the range query from 1.00 shard to 4.00 shards; the weighted average climbed from 2.95 to 3.40. - A compound key depends on the prefix condition: the `{branch,member_id}` placement balanced perfectly, but the member query, which does not pin its prefix, scattered across 4.00 shards and came out worst on the weighted average, 3.50. - In this mix, the best key was `member_id`: 1.00 times balance, a 27 percent hot share, and 2.65 weighted shards. The choice is made not by looking at balance but at the mix's heaviest query. ## Next Step This lesson resolved which shard a document goes to, but the shard itself is the previous lesson's replica set: every shard is made up of more than one member, and which of those members has to have seen a write is still open. Both lessons left the same gap — in the first lesson, 6 accepted loan records were rolled back, and how to prevent that was never said. How many members have to acknowledge a write for it to count as "successful" is a setting; in the same way, which member a read is answered from and with what verification is also a setting. The next lesson names these two settings and runs the same loan scenario under three of them: the number of turns an acknowledgement takes to arrive, the number of writes lost at the moment of failure, and how the version seen on a read changes.