Sharding and Replication

Questions to answer:

what is sharding and replication, what are they for, and why a real system usually does both
how to distribute (shard) data among nodes in a cluster
what is the cost of sharding
what is the cost of replication (lag)
how are nodes kept synchronized (eventual consistency)
how Redis Cluster, MongoDB and Neo4j distribute data

Why is one machine not enough

Recall the two ways to scale:

vertical (a more powerful server)

simple, but at some point the next bigger server is too expensive or unavailable
horizontal (more servers)

a cluster of ordinary machines (nodes) that cooperate

Two questions for a piece of data in a cluster:
- Which machine has it? → sharding algorithm
- What if a node is not accessible? → replication

Sharding cuts the dataset into pieces and puts them on different nodes.

→ increased capacity: larger data store, writes to different keys/machines can run concurrently
Replication puts complete copies of the same data on several nodes.

→ better dependability: data does not disappear with a dead node
→ higher read throughput: multiple nodes can answer reads in parallel

Shard/partition key — decides where a record goes (placement function) in the cluster
Shard/partition — the set of records that share the placement (node)
Router — maps/forwards requests: key X ⇒ node Y
Redis: a client library with a cached map. MongoDB: a proxy process mongos.

Range: keys are sorted and divided into intervals; each interval maps to one node.
- cheap, neighbours stay together
- problem: if a key always grows (timestamp, auto-increment id) → all writes to the last interval/node — the hot tail
Hashing: a hash of the key maps to the node.
- consecutive keys scatter, writes spread evenly even for growing keys
- problem: a range scan — the router must ask many nodes and merge results (scatter–gather)

Consistent hashing: nodes and keys form a ring. Key is placed on a node as hash(key) mod n, cheaper to add and remove nodes 🤔 Consistent hashing^🌐

Scatter-gather: a query without a partition key must be sent to every shard and merged; more shards make such queries slower
Cross-shard operations: joins or atomic updates of two+ records are easy on one node, hard across shards → design: one request, one shard
Rebalancing: when a node joins or leaves, or the load changes, data must move while the cluster keeps answering
Data dies with its node: one copy of a shard = a single point of failure

Redis Cluster: 16 384 hash slots, CRC16(key) mod 16384; a map stores which primary owns which slot
MongoDB: arbitrary shard keys, documents grouped into chunks, a chunk map maps chunks to shards, a balancer moves chunks among shards.

The cost: replication lag — a read from a lagging secondary returns an older value (inconsistency window}

They use primary–secondary replication.

A problem: a client that writes and then reads from a secondary can miss its own write. (There are ways to fix it, but they have a cost.)

The secondaries notice and elect a new primary.

A network partition that leaves two groups → two primaries that accept writes and create conflicting histories. Must be prevented.

Three examples:

Session cache (Redis): one small record per logged-in user. Replicate so one dead node does not log everyone out; shard by user id, once there are many users.
Order history (MongoDB): a large collection, kept for years. Shard by customer id; a daily report is a scatter–gather. Replicate each shard, because an acknowledged order must survive.
Social graph (Neo4j): Friends-of-friends. The graph fits on one machine. Replicate: one leader accepts writes, secondaries serve traversals.

The decisions:

Optimize sharding for requests that must be fast: Everything else can be a scatter–gather operation.
Choose how many copies confirm a write: One copy is fast but fragile; every copy means one slow replica fails the write; a majority is the usual default.
Decide whether reads may use secondaries: Yes: scales reads and risks stale data. No: guarantees fresh data but burdens the primary.