Sharding

Federation split the store’s work, but each context’s own data still sits whole on a single node. The delivery operations store carries 624.88 GB and takes 250.00 operations per second; these numbers do not shrink by splitting the context further, because there is no context left to split. The one path left is to split the data within itself.

Sharding is distributing the same context’s records across independent nodes by a key. It should not be confused with the partitioning from the Relational Database Administration course: partitioning divides a table into pieces inside the same engine, while sharding distributes those pieces across separate engines and machines. That course counted the costs sharding imposes on the application — joins, uniqueness and transaction boundaries, the scatter-gather pattern breaking the tail, the mapping method setting the move ratio on rebalancing — and they are not retold here. This lesson’s question is a scaling decision: which key is chosen, and what the choice does to the access patterns.

Key Candidates and Their Distributions

The shipment data has three candidates, and all three are the key of a real question: tracking number (P1’s key), carrier (P2’s key), and seller (P3’s key). Two properties separate the candidates — how many distinct values they take, and how evenly those values are distributed.

VD6 — the distribution of the three candidate keys. Tracking number is uniform and distinct for every shipment; carrier takes 12 distinct values and the largest carrier’s share is 0.35; seller takes 4,000 distinct values and the largest 1-percent slice (40 sellers) produces 0.30 of shipments. Rationale: tracking number is a generated identifier; carrier traffic splits among a small number of large players; seller volume concentrates in a small number of large sellers. The seller count comes from K01’s arithmetic (4,000 invoice lines a day = 4,000 sellers); the rest is this lesson’s assumption and is not added to K01’s table. Sensitivity is given below at a carrier share of 0.50.

Mapping a key to a node uses consistent hashing. The rule itself, and the fraction of keys that move when the node count changes, were measured in the Traffic Layer course; here it is used only for data placement, and that measurement is not repeated.

// sharding/key.mjs — the data distribution, hot spot, touched nodes, and rebalancing cost of
// three candidate shard keys. The placement rule is consistent hashing; the rule itself and the
// move ratio were measured in M19/K02, used here only for data placement. THIS IS A MODEL.
const SHIPMENTS = 100_000, SHARDS = 8, VIRTUAL = 200;   // model parameters
const CARRIER = 12, CARRIER_SHARE = 0.35;               // VD6
const SELLER = 4000, BIG = 40, BIG_SHARE = 0.30;        // VD6
let s = 20260731 % 2147483647;
const rand = () => (s = (s * 48271) % 2147483647) / 2147483647;

function hash(text) {                                   // FNV-1a; the function was set up in M01/K03
  let h = 2166136261;
  for (let i = 0; i < text.length; i += 1) h = Math.imul(h ^ text.charCodeAt(i), 16777619) >>> 0;
  h ^= h >>> 15; h = Math.imul(h, 2246822507) >>> 0; h ^= h >>> 13;
  return h >>> 0;
}
const ring = (n) => {
  const nodes = [];
  for (let d = 0; d < n; d += 1) for (let v = 0; v < VIRTUAL; v += 1) nodes.push([hash(`d${d}#${v}`), d]);
  return nodes.sort((a, b) => a[0] - b[0]);
};
const place = (key, r) => {
  const c = hash(key);
  if (c > r[r.length - 1][0]) return r[0][1];
  let lo = 0, hi = r.length - 1;
  while (lo < hi) { const m = (lo + hi) >> 1; if (r[m][0] < c) lo = m + 1; else hi = m; }
  return r[lo][1];
};

const records = [];
for (let i = 0; i < SHIPMENTS; i += 1) records.push({
  trackingNo: `TR-${1_000_000 + i}`,
  carrier: `T${rand() < CARRIER_SHARE ? 0 : 1 + Math.floor(rand() * (CARRIER - 1))}`,
  seller: `S${rand() < BIG_SHARE ? Math.floor(rand() * BIG) : BIG + Math.floor(rand() * (SELLER - BIG))}`,
});

const CANDIDATES = ["trackingNo", "carrier", "seller"];
const ring8 = ring(SHARDS), ring9 = ring(SHARDS + 1);
const READ = 41.67, WRITE = 97.22;                    // K01: requests/s reaching the store for P1 and P2
const GB = 624.88, DAILY_MB = 856;                    // lesson 03: delivery operations store
const TOUCHED = { trackingNo: [1, 1, SHARDS], carrier: [SHARDS, 1, SHARDS], seller: [SHARDS, SHARDS, 1] };
const b = (x, n = 2) => x.toFixed(n);

console.log(`model: ${SHIPMENTS.toLocaleString("en-US")} shipments, ${SHARDS} shards, ${VIRTUAL} virtual nodes`);
console.log(`\n${"shard key".padEnd(16)}${"distinct keys".padStart(15)}${"hot spot".padStart(12)}` +
  `${"busiest GB".padStart(13)}${"busiest ops/s".padStart(16)}${"8->9 moved".padStart(13)}${"moved GB".padStart(11)}`);
for (const a of CANDIDATES) {
  const y8 = records.map((k) => place(k[a], ring8)), y9 = records.map((k) => place(k[a], ring9));
  const counts = new Array(SHARDS).fill(0);
  for (const d of y8) counts[d] += 1;
  const busiest = Math.max(...counts) / SHIPMENTS, hotspot = busiest * SHARDS;
  const moved = y8.filter((d, i) => d !== y9[i]).length / SHIPMENTS;
  console.log(`${a.padEnd(16)}${String(new Set(records.map((k) => k[a])).size).padStart(15)}` +
    `${b(hotspot, 3).padStart(12)}${b(GB * busiest).padStart(13)}` +
    `${b((READ + WRITE) * busiest).padStart(16)}${b(moved, 4).padStart(13)}${b(GB * moved).padStart(11)}`);
}
console.log(`perfect distribution has hot spot 1.000; at ${SHARDS} shards, per node ` +
  `${b(GB / SHARDS)} GB and ${b((READ + WRITE) / SHARDS)} ops/s`);
{
  const y8 = records.map((k) => place(k.trackingNo, ring8)), y9 = records.map((k) => place(k.trackingNo, ring9));
  const moved = y8.filter((d, i) => d !== y9[i]).length / SHIPMENTS, movedGB = GB * moved * 1000;
  console.log(`moved with tracking number as key: ${b(movedGB / 1000)} GB = ` +
    `${b(movedGB / DAILY_MB, 1)} days of the delivery store's daily growth (lesson 03: ${DAILY_MB} MB/day)`);
}

console.log(`\n${"shard key".padEnd(16)}${"P1 nodes".padStart(10)}${"P2 nodes".padStart(10)}` +
  `${"P3 nodes".padStart(10)}${"online touches/s".padStart(19)}${"baseline factor".padStart(17)}`);
for (const a of CANDIDATES) {
  const [p1, p2, p3] = TOUCHED[a], touches = READ * p1 + WRITE * p2;
  console.log(`${a.padEnd(16)}${String(p1).padStart(10)}${String(p2).padStart(10)}${String(p3).padStart(10)}` +
    `${b(touches).padStart(19)}${b(touches / (READ + WRITE)).padStart(17)}`);
}
console.log(`online touches in the unsharded store ${b(READ + WRITE)} /s = K01's requests/s reaching the store`);

// VD6 sensitivity: if the largest carrier's share were 0.50 instead of 0.35 (carrier field regenerated)
{
  const counts = new Array(SHARDS).fill(0);
  for (const k of records)
    counts[place(`T${rand() < 0.5 ? 0 : 1 + Math.floor(rand() * (CARRIER - 1))}`, ring8)] += 1;
  const busiest = Math.max(...counts) / SHIPMENTS;
  console.log(`\nsensitivity: if the largest carrier's share were 0.50, hot spot would be ` +
    `${b(busiest * SHARDS, 3)} and the busiest node ${b(GB * busiest)} GB (in the table, at 0.35: ` +
    `${b(3.748, 3)} and ${b(292.78)} GB)`);
}

model: 100,000 shipments, 8 shards, 200 virtual nodes

shard key         distinct keys    hot spot   busiest GB   busiest ops/s   8->9 moved   moved GB
trackingNo               100000       1.116        87.14           19.37       0.1193      74.54
carrier                      12       3.748       292.78           65.07       0.1182      73.85
seller                     4000       1.148        89.63           19.92       0.1100      68.74
perfect distribution has hot spot 1.000; at 8 shards, per node 78.11 GB and 17.36 ops/s
moved with tracking number as key: 74.54 GB = 87.1 days of the delivery store's daily growth (lesson 03: 856 MB/day)

shard key         P1 nodes  P2 nodes  P3 nodes   online touches/s  baseline factor
trackingNo               1         1         8             138.89             1.00
carrier                  8         1         8             430.58             3.10
seller                   8         8         1            1111.12             8.00
online touches in the unsharded store 138.89 /s = K01's requests/s reaching the store

sensitivity: if the largest carrier's share were 0.50, hot spot would be 4.713 and the busiest node 368.15 GB (in the table, at 0.35: 3.748 and 292.78 GB)

These numbers belong to the calculation class: they are deterministic, come from a fixed-seed generator and a fixed placement rule, and do not depend on the machine.

The Hot Spot Depends on How Many Values the Key Takes

The first table’s hot spot column separates the three candidates immediately. Hot spot is the busiest node’s share divided by the perfect share; 1.000 is a perfect distribution.

Tracking number gives 1.116 and seller gives 1.148; both are acceptable. Carrier gives 3.748: the busiest node carries almost four times the perfect share, 292.78 GB and 65.07 operations per second. In an eight-node cluster, 78.11 GB and 17.36 ops/s are expected per node, and one node takes four times that — for that node, sharding might as well not have happened.

The reason is how many distinct values the key takes. Carrier takes only 12 values, and the largest one alone carries 0.35 of the load; no placement rule can split a single key across two nodes, because the key is the unit of indivisibility. Seller is skewed too — 40 sellers produce 0.30 of volume — but spread across 4,000 distinct values, the skew dissolves in placement: the largest seller’s own share is 0.0075 and cannot capture a single node by itself. The rule is this: what creates a hot spot is not skew, but skew concentrated in a small number of key values. The sensitivity row confirms it — at a carrier share of 0.50, the hot spot would climb to 4.713 and the busiest node to 368.15 GB.

The Number of Nodes Touched Changes with the Pattern

The second table flips the choice around. If a request cannot filter on the shard key, it has to go to every node.

With tracking number as the key, P1 and P2 each touch one node: the tracking query already knows the number, and the state event carries it too. The online touch rate stays at 138.89 node-touches/s — K01’s requests reaching store/s number itself, since sharding adds no extra touch to the online path. With carrier as the key, P1 cannot tell the carrier from the number, so it asks all eight nodes at once, and the rate climbs to 430.58, a 3.10× factor over the baseline. With seller as the key, both P1 and P2 go to all eight nodes: 1,111.12 touches/s, exactly 8.00× the baseline.

The third pattern completes the table. P3 stays on a single node only with seller as the key; with the others, it spreads across all eight. But P3 runs once a day with a four-hour window, so spreading across eight nodes is not a latency problem for it, only a coordination task. P1 and P2, by contrast, flow continuously at 138.89 requests a second, and tracking reads carry a 200-millisecond threshold.

The decision comes out of where these two tables intersect, and it fits one sentence: the shard key is the field the most frequent, tightest-threshold pattern filters on. Tracking number wins on both tables — hot spot of 1.116, online touch at the baseline. Carrier keeps P2 on a single node but spreads P1 across eight and adds a 3.748 hot spot on top of that; seller saves P3 but multiplies the online path by eight.

The Cost of Rebalancing

The last two columns give what happens when a node is added. Going from eight shards to nine, 0.1193 of keys move under tracking number as the key — close to the theoretical lower bound of 1/9, or 0.1111. The ratio itself was what the Traffic Layer course measured; here it converts to bytes: 74.54 GB out of a 624.88 GB store.

The size alone means nothing; it needs a yardstick. The delivery operations store grows 856 MB a day (lesson 03), so 74.54 GB equals 87.1 days of that store’s growth — adding one node means moving roughly three months of growth over the network all over again. That is why rebalancing is a planned job done without downtime, claiming its own share of bandwidth — not an instant configuration change but a data-moving job.

With carrier as the key, the ratio is similar (0.1182), but its meaning changes: when one of 12 keys moves, all of that key’s data moves at once, and during the move two nodes together carry a large share of the load. That is the second cost of a key with few distinct values.

Summary

• Sharding distributes the same context’s records across separate engines; table partitioning stays inside the same engine, and the two are separate decisions.
• A hot spot is created not by skew but by skew concentrated in a small number of key values: a hot spot of 3.748 (292.78 GB, 65.07 ops/s) for the 12-valued carrier, 1.148 for the 4,000-valued seller, 1.116 for the uniform tracking number.
• A pattern that cannot filter on the shard key goes to every node: online touch rate is 138.89/s (baseline) with tracking number as the key, 430.58 (3.10×) with carrier, 1,111.12 (8.00×) with seller.
• The key is the field the most frequent, tightest-threshold pattern filters on; a batch scan with a four-hour window can absorb the scatter-gather cost, an online path running at 138.89 requests a second cannot.
• At eight shards, per-node load is 78.11 GB and 17.36 ops/s; at VD6’s carrier share of 0.50, the busiest node would climb to 368.15 GB.
• Going from eight shards to nine moves 0.1193 of keys, equal to 74.54 GB — 87.1 days of the delivery operations store’s growth.

Next Step

The sharding decision is made and the key is chosen: tracking number. But one question was left open, waved off with a single answer in this lesson — by what rule the key gets distributed across shards. Consistent hashing was used here because it had been measured in the Traffic Layer course, but it is not the only option. A key can be split by range, distributed by its hash, or mapped from key to shard through a separate lookup table. All three place the same data on the same nodes in different shapes, and all three affect rebalancing differently: under one, a range scan stays on a single node but new records keep landing on the same node; under another, the distribution evens out but the range scan spreads out too; under the third, the mapping can be changed freely but every lookup passes through a layer of indirection. The next lesson compares these three strategies.