Partitioning, Sharding, and Replication

"They split me into eight pieces so I would fit, then copied each piece three times so I would survive. I have never been so divided and so redundant at once."

A Shard That Believes It Is the Whole Model

The previous section established that a distributed system is a collection of machines coordinating through messages, and that no single machine holds the whole picture. This section asks the immediate next question: given that the state must live on many machines, how do we decide which byte goes where? Two answers cover the entire design space. We can divide the state into disjoint pieces and scatter the pieces, which we call partitioning or sharding, or we can copy the state and place the copies on multiple machines, which we call replication. Real systems do both at once, and the interesting engineering is in choosing the partition function, the replication factor, and the consistency rules that hold the copies together. We take the two moves in turn, then show how they compose, and close by tracing where each one resurfaces in the AI chapters ahead.

1. Partitioning: Cutting One Dataset Into Disjoint Shards Beginner

Partitioning, also called sharding, splits a single logical dataset or structure into disjoint subsets and assigns each subset to a different node. The word shard pictures it well: one pane of glass broken into pieces that together make up the whole, with no piece overlapping another. A partition function maps each item to exactly one shard, so a key of interest has a single home and a reader knows where to look. The payoff is twofold. Capacity grows with the number of nodes, because the data that would not fit on one machine now occupies many. Throughput grows too, because independent shards can be scanned, updated, or trained on in parallel without coordinating. The cost is that any operation spanning many keys, a join, a global aggregation, a nearest-neighbor search, must now touch many shards and combine their answers, which is exactly the communication tax that Chapter 3 teaches you to price.

Two partition functions dominate practice, and they trade the same two properties against each other: how evenly load is spread, and whether items that are close in key space land near each other. Hash partitioning sends item with key $k$ to shard $\text{hash}(k) \bmod M$ for $M$ shards. A good hash scatters keys uniformly, so each shard receives roughly $1/M$ of the items regardless of how skewed the original keys were; this is the default when you want balanced load and only ever look keys up one at a time. Range partitioning instead assigns contiguous key intervals to shards: keys $[a, b)$ to shard 1, $[b, c)$ to shard 2, and so on. Range partitioning keeps neighboring keys together, which makes range scans and ordered iteration cheap, but it invites hot spots when traffic concentrates in one interval (think of timestamps, where today's range absorbs every new write). Table 2.3.1 lays the trade-off out directly.

2. Consistent Hashing: Rebalancing When Nodes Join and Leave Intermediate

Plain hash partitioning has a hidden flaw that surfaces the moment the cluster changes size. The shard of key $k$ is $\text{hash}(k) \bmod M$, and that formula depends on $M$. Add one node so $M$ goes from 8 to 9, and almost every key now hashes to a different shard, because changing the modulus reshuffles the entire mapping. In a static cluster this never matters, but real clusters grow, shrink, and replace failed machines constantly, and re-homing nearly all the data every time a single node joins is ruinous: it saturates the network, evicts caches, and stalls the workload for the duration of the move. We want the opposite property, that adding the $(M{+}1)$-th node moves only about $1/(M{+}1)$ of the keys, the bare minimum needed to give the newcomer its fair share.

Consistent hashing delivers exactly that property. Picture the hash output space as a ring (the integers modulo $2^{32}$, wrapped end to end). Each node is hashed to one or more positions on the ring, and each key is hashed to a position too; a key belongs to the first node found by walking clockwise from the key's position. The crucial consequence is locality of change: when a node joins, it lands at some point on the ring and takes over only the keys in the arc between it and its clockwise neighbor. Every other key keeps its old home untouched. To keep load even and avoid a single node owning a long arc by bad luck, each physical node is placed at many positions, called virtual nodes; the more virtual nodes, the smoother the load distribution and the closer the moved fraction gets to the ideal $1/(M{+}1)$. The demonstration below builds both schemes on the same 100,000 keys, adds one node to an 8-node cluster, and counts how many keys change homes.

import hashlib, bisect

def h(s):                                     # stable hash, independent of run
    return int(hashlib.md5(str(s).encode()).hexdigest(), 16)

NUM_KEYS = 100_000
keys = [f"key-{i}" for i in range(NUM_KEYS)]

# Plain hash (modulo) partitioning: shard = hash(k) % M
def mod_assign(M):
    return {k: h(k) % M for k in keys}

before_mod = mod_assign(8)
after_mod  = mod_assign(9)                    # one node joins: M goes 8 -> 9
moved_mod  = sum(1 for k in keys if before_mod[k] != after_mod[k])

# Consistent hashing on a ring, with virtual nodes for smooth load
RING   = 1 << 32
VNODES = 200                                  # replicas per node smooth the ring

def build_ring(node_ids):
    ring = sorted((h(f"node-{n}-vn-{v}") % RING, n)
                  for n in node_ids for v in range(VNODES))
    return [p for p, _ in ring], [n for _, n in ring]

def ch_assign(node_ids):
    pos, owners = build_ring(node_ids)
    return {k: owners[bisect.bisect(pos, h(k) % RING) % len(pos)] for k in keys}

before_ch = ch_assign(range(8))
after_ch  = ch_assign(range(9))               # add node 8 to the ring
moved_ch  = sum(1 for k in keys if before_ch[k] != after_ch[k])

print(f"keys total                    : {NUM_KEYS}")
print(f"nodes before / after          : 8 -> 9 (one node joins)")
print(f"plain hash mod  keys moved     : {moved_mod:>7}  ({100*moved_mod/NUM_KEYS:5.1f}% of all keys)")
print(f"consistent hash keys moved     : {moved_ch:>7}  ({100*moved_ch/NUM_KEYS:5.1f}% of all keys)")
print(f"ideal minimum (1/9 of keys)    : {NUM_KEYS//9:>7}  ({100/9:5.1f}%)")
print(f"reduction factor               : {moved_mod/moved_ch:5.1f}x fewer keys moved")

keys total                    : 100000
nodes before / after          : 8 -> 9 (one node joins)
plain hash mod  keys moved     :   88863  ( 88.9% of all keys)
consistent hash keys moved     :   11280  ( 11.3% of all keys)
ideal minimum (1/9 of keys)    :   11111  ( 11.1%)
reduction factor               :   7.9x fewer keys moved

The numbers tell the whole story. Plain modulo hashing scrambled nearly nine tenths of the keys to absorb one new machine, while consistent hashing moved only the keys the new machine had to take, a touch over one ninth. That gap is the difference between a cluster that can grow and shrink gracefully and one that seizes up every time its membership changes. This is why consistent hashing (introduced for distributed web caches and popularized by the Dynamo storage system) underlies the partitioning layer of countless distributed databases, caches, and, as we will see, the embedding and vector stores that AI systems lean on.

3. Replication: Copying State for Availability and Read Throughput Beginner

Partitioning answers "how do we fit and parallelize?" but it does nothing for survival: if the single node holding shard 3 dies, shard 3 is gone, and any operation needing those keys fails. Replication answers the survival question by copying state onto several nodes. With a replication factor of three, every piece of data lives on three machines, so the system tolerates the loss of two of them without losing the data, and three readers can be served at once from the three copies. Replication is the source of the high availability that production systems advertise, and it is the reason a well-built service stays up while individual machines underneath it crash, reboot, and get replaced. The two motives reinforce each other: copies provide both fault tolerance (a copy survives the loss of the original) and read scaling (many copies share the read load).

The catch is that copies must agree. The instant you have more than one copy of a mutable value, a write has to reach all of them, and until it does, the copies disagree: a reader hitting a stale copy sees an old value. This is the consistency cost of replication, and it is not a bug to be fixed but a permanent tension to be managed. You can insist that every write reach every copy before it is acknowledged, which keeps copies identical but makes writes slow and blocks them entirely when a copy is unreachable. Or you can let writes return after reaching some copies and propagate to the rest in the background, which keeps writes fast and available but lets readers occasionally see stale data, a posture called eventual consistency. The synchronous-versus-asynchronous choice you meet here returns throughout the book, as synchronous versus asynchronous gradient updates in Chapter 10 and as bounded-staleness parameter reads in Chapter 11. The deeper impossibility result that forces this trade-off, that a replicated system cannot be simultaneously consistent and available when the network partitions, is the CAP theorem, formalized in Section 2.5.

4. Composing the Two: Partition for Capacity, Replicate for Safety Intermediate

Production systems never choose between partitioning and replication; they layer them. First partition the state so it fits and parallelizes, then replicate each partition so the loss of a node does not lose a shard. A dataset of one petabyte across 100 nodes might be cut into 100 shards of ten terabytes each, with each shard copied onto three nodes, giving 300 storage assignments that survive any two simultaneous failures while still letting 100 shards be read in parallel. The replication factor controls how many failures you tolerate; the shard count controls how much you parallelize; the partition function controls how evenly load lands. These three knobs, set independently, describe essentially every distributed storage and state layer you will meet, including the ones built specifically for AI.

Composition also inherits the costs of both moves. Partitioning's cost is that multi-shard operations must gather from many places; replication's cost is that every write must fan out to several copies and the copies can disagree. A system that both shards and replicates pays both taxes, which is why the cheapest distributed operation is always the one that touches a single shard's single up-to-date copy, and why so much distributed-systems engineering is about arranging for the common case to be exactly that. Keeping these costs explicit, rather than hidden behind a convenient abstraction, is the habit this chapter is trying to build, and Chapter 6 shows the first large payoff when MapReduce turns partitioned data into parallel computation.

5. Where Partitioning and Replication Return in AI Intermediate

The two moves are not background infrastructure that AI happens to sit on; they are load-bearing in the AI methods themselves, and the same vocabulary recurs at every layer of the stack. The clearest case is the training data. A web-scale corpus is partitioned into data shards so that many workers each stream a disjoint slice in parallel, which is precisely the data-parallel setup from Section 2.2 seen from the storage side; Chapter 8 builds the distributed loaders that feed those shards to GPUs without starving them, and a replication factor on the underlying object store keeps a shard available when a storage node fails.

Partitioning reappears, far less obviously, inside the model. When an embedding table or a parameter vector is too large for one accelerator, it is sharded across devices exactly like a dataset: Chapter 11 shards billion-row embedding tables across parameter servers using the hash partitioning of this section, and Chapter 16 shards the parameters, gradients, and optimizer state of a single giant model across GPUs in the ZeRO and FSDP schemes, where the "shard" is a slice of one weight tensor rather than a slice of data. At serving time the pattern returns once more: a vector index too large to search on one node is split into index shards that are searched in parallel and whose partial results are merged, the design Chapter 25 develops for distributed retrieval and approximate nearest-neighbor search. Table 2.3.2 collects these recurrences so you can see one idea wearing many costumes.

from uhashring import HashRing                 # pip install uhashring

ring = HashRing(nodes=[f"node-{i}" for i in range(8)])   # 8 nodes, vnodes by default
home = ring.get_node("key-42")                # which node owns this key
ring.add_node("node-8")                        # join: only ~1/9 of keys re-home
ring.remove_node("node-3")                     # leave: only that node's keys move

6. The Frontier: Sharding the Largest Models and Indexes Advanced

Partitioning and replication are old ideas, but the scale of modern AI keeps forcing new variants of them, and the 2024 to 2026 literature is full of sharding strategies invented because the thing being split got too large for the previous one. On the training side, the FSDP and ZeRO family (Zhao et al., 2023, PyTorch FSDP; and the ongoing ZeR++ and per-parameter sharding work in DeepSpeed through 2024) shard not just data but the optimizer state, gradients, and parameters of a single model across hundreds of GPUs, then re-gather each shard just in time for the layer that needs it, treating a weight tensor exactly as this section treats a dataset. On the serving side, distributed vector search has become a research field of its own: systems in the lineage of DiskANN and the 2024 wave of disaggregated and GPU-accelerated vector databases (such as the partitioned-index designs surveyed for billion-scale retrieval-augmented generation) shard a billion-vector index across machines and tune the partition function to keep each query touching as few shards as possible, because every extra shard a query fans out to is added tail latency. The common research question across both is sharper than the classic one: not merely how to split state, but how to split it so that the recombination, the all-gather of weights or the merge of partial search results, stays cheap as the shard count climbs. We meet the training answer in Chapter 16 and the serving answer in Chapter 25.

We now have the third concept of this chapter in hand: state lives on many machines either by partitioning (disjoint shards for capacity and parallelism, with consistent hashing to rebalance gracefully) or by replication (copies for availability and read throughput, at the cost of keeping them consistent), and real systems compose the two. What we have not yet confronted is what happens when the machines holding those shards and copies fail, which is not an exceptional event but the normal condition of a large cluster. Turning replication from a throughput trick into a survival strategy, and deciding what "still correct" means when copies disagree during a failure, is the subject of Section 2.4.