Sharding

Fullscreen

Terminology zoo: a "shard" is called a partition (Kafka), range (CockroachDB), region (HBase, TiDB), vBucket (Couchbase), vnode/token-range (Riak/Cassandra), or tablet (Bigtable, YugabyteDB, ScyllaDB). In PostgreSQL, partitioning = splitting a table into files on one machine, while sharding = splitting across machines; elsewhere the words are synonyms. Sharding has nothing to do with network partitions (netsplits, Chapter 9).

Fullscreen

---

Pros and cons of sharding

• Primary reason: scalability, the main tool for horizontal scaling (scale-out / shared-nothing, Chapter 2). If each shard takes a roughly equal share, 10 nodes handle ~10× the data and throughput. (If only read throughput is the problem, use read replicas instead, Chapter 6.)
• Replication is useful at any scale; sharding is a heavyweight, large-scale tool. If one machine can handle your data + writes (and modern machines can handle a lot), avoid sharding.
• Costs of sharding:
  • You must pick a partition key; all records with the same partition key go to the same shard. Access is fast if you know the shard, but otherwise you must search all shards. The scheme is hard to change.
  • Works well for key-value data; harder for relational data (secondary-index searches, joins across shards).
  • A write touching related records in several shards needs a distributed transaction (Chapter 8), much slower than single-node, often a bottleneck.
• Even on one machine: Redis, VoltDB, FoundationDB run one single-threaded process per CPU core, using sharding to exploit cores / NUMA.

Sharding for multitenancy

SaaS products are often multitenant (each customer = a tenant with a self-contained dataset). Give each tenant its own shard (or group small tenants together). Advantages:

Challenges: a single tenant may be too big for one node (needs sharding within the tenant); many tiny tenants create per-shard overhead (and moving tenants between shards as they grow is hard); cross-tenant features require cross-shard joins.

---

Sharding of Key-Value Data

Goal: spread data and query load evenly. Imbalance = skew; a shard with disproportionate load = a hot shard / hot spot; a single high-load key (a celebrity) = a hot key. You need an algorithm mapping a partition key → shard that's also amenable to rebalancing.

Sharding by key range

Assign each shard a contiguous range of keys (like volumes of an encyclopedia).

Fullscreen

• Ranges are not evenly spaced, boundaries adapt to the data distribution (chosen manually, e.g. Vitess, or automatically, e.g. Bigtable/HBase, CockroachDB, MongoDB range mode, FoundationDB).
• Keys are stored sorted within a shard (B-tree/SSTables) → range scans are easy, and you can treat the key as a concatenated index (e.g., fetch all sensor readings for a month).
• Downside, hot spots from sequential writes: if the key is a timestamp, all current writes hit one shard (this month). Fix: prefix the key with something like a sensor ID so writes spread out, at the cost of needing a separate range query per sensor.
• Rebalancing: start with pre-splitting (configure initial shards if you know the distribution); grow by splitting an over-large/over-loaded shard into subranges (default trigger e.g. 10 GB in HBase, or sustained high write throughput); merge adjacent small shards. The number of shards adapts to data volume. Splitting is expensive (rewrites all the shard's data, like an LSM compaction) and often hits an already-hot shard.

Sharding by hash of key

If you don't need nearby keys grouped (e.g., tenant IDs), hash the partition key first, a good hash turns skewed input into a uniform distribution. Needn't be cryptographic (MongoDB uses MD5; Cassandra/ScyllaDB use Murmur3). Avoid language built-ins like Java's hashCode() (inconsistent across processes).

The mod-N trap: assigning hash(key) % N nodes seems easy, but changing N moves most keys.

Fullscreen

Fix 1, fixed number of shards: create many more shards than nodes (e.g., 1,000 shards on 10 nodes), and assign several shards per node.

Fullscreen

• Only whole shards move (cheaper than splitting); shard count and key→shard mapping stay fixed; only shard→node changes. Old mapping serves reads/writes during transfer.
• Choose a count divisible by many factors; give powerful nodes more shards.
• Used by Citus, Riak, Elasticsearch, Couchbase. Limits: you can't have more nodes than shards; if you picked the wrong count, resharding is expensive (split every shard); shard size grows with the dataset, so a fixed count is awkward when total size varies a lot ("just right" shard size is hard to hit).

Fix 2, sharding by hash range: assign each shard a range of hash values (e.g., a 16-bit hash 0-65,535 split into ranges).

• Number of shards adapts (split when too big/loaded), no need to predict in advance.
• Range queries on the partition key are inefficient (nearby keys scatter), but if the key has multiple columns and only the first is the partition key, range queries over later columns still work (same partition key → same shard).
• Used by YugabyteDB, DynamoDB, and optionally MongoDB. Cassandra/ScyllaDB variant: split the hash space into ranges proportional to node count with random boundaries (16 ranges/node default in Cassandra, 256 in ScyllaDB); multiple ranges per node even out imbalances, and adding a node takes parts of several existing ranges (fair share, minimal data moved).

Warehouses do similar things: BigQuery (partition key + "cluster columns"), Snowflake ("micro-partitions" + cluster keys), Delta Lake, clustering improves range scans, compression, and filtering.

Consistent hashing

A consistent hashing algorithm maps keys to a chosen number of shards so that (1) keys spread roughly equally, and (2) when the shard count changes, as few keys move as possible. ("Consistent" here is unrelated to replica or ACID consistency.)

Fullscreen

Cassandra/ScyllaDB's scheme resembles the original; other algorithms include rendezvous (highest random weight) hashing and jump consistent hashing. Some split existing shards for a new node; others assign the new node individual keys scattered from all nodes.

Skewed workloads and relieving hot spots

Consistent hashing spreads keys evenly, but not necessarily load. A celebrity post can hammer one key (one partition key). Mitigations:

Fullscreen

• Range-based schemes can isolate a hot key in its own shard / dedicated machine.
• Key salting: append a random suffix to spread a hot key's writes across many keys, but reads must now gather from all of them (read volume isn't reduced), and you need bookkeeping for which keys are split (only worth it for the few hot keys). Load also shifts over time (a viral post cools down), and read-hot vs write-hot keys need different strategies.
• Cloud systems automate this: Amazon's heat management / adaptive capacity.

Automatic vs manual rebalancing

• Fully automatic: less ops work, can autoscale (DynamoDB adapts in minutes). But rebalancing is expensive (reroute + move lots of data); done carelessly it overloads the network/nodes, and near max throughput the split may not keep up.
• Danger with auto failure-detection: an overloaded (slow) node is mistaken for dead → cluster rebalances away from it → more load → cascading failure.
• Middle ground: systems (Couchbase, Riak) suggest an assignment but require an admin to commit it. A human in the loop is slower but avoids surprises, and lets you pre-balance before known surges (Cyber Monday, World Cup tickets).

---

Request Routing

If you want a key, which node (IP + port) do you connect to? This is request routing, similar to service discovery, but a key can only be served by a node that replicates that key's shard (not just any stateless instance).

Fullscreen

1. Any node, client hits any node (round-robin); it serves or forwards.
2. Routing tier, a shard-aware load balancer forwards to the owner.
3. Shard-aware client, connects directly to the owner.

Key problems: who decides shard→node assignment (a coordinator, but make it fault-tolerant and split-brain-safe)? How does the router learn of changes? What about in-flight requests during a cutover?

• Many systems use a coordination service (ZooKeeper / etcd) with a consensus algorithm (Chapter 10) as the authoritative shard→node map; nodes register, routers subscribe and get notified on changes (HBase, SolrCloud use ZooKeeper; Kubernetes uses etcd; MongoDB uses its own config servers + mongos; Kafka, YugabyteDB, TiDB, ScyllaDB embed Raft).
• Riak uses a gossip protocol instead, weaker consistency (split brain possible), acceptable for leaderless DBs with weak guarantees.
• DNS is usually enough for finding the (slowly changing) node IPs.
• Analytical databases shard too, but a query typically aggregates/joins across many shards in parallel (Chapter 11), unlike single-shard OLTP lookups.

---

Sharding and Secondary Indexes

Everything so far assumed you know the partition key. Secondary indexes (find all actions by user 123, all red cars) don't map neatly to shards. Two approaches:

Fullscreen

Local secondary indexes (document-partitioned)

Each shard indexes only its own records. Writes touch just one shard (simple, in sync). But a query by secondary value must scatter to all shards and gather results (matching records can be anywhere), this is a scatter/gather read, prone to tail latency amplification, and adding shards doesn't raise query throughput (every shard processes every query). Despite this, widely used: MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud, VoltDB.

Global secondary indexes (term-partitioned)

Build one index covering all shards, but shard the index itself by the indexed term (e.g., colors a-r in index shard 0, s-z in shard 1).

• Reads: a single-condition query (color = red) reads the postings list from one shard. (Fetching the actual records still needs the shards holding those IDs; multi-term AND queries may need to intersect postings lists across shards.)
• Writes are harder: one record's terms may live on many index shards → keeping the index in sync is complex; one option is a distributed transaction (Chapter 8).
• Used by CockroachDB, TiDB, YugabyteDB; DynamoDB supports both local and global. DynamoDB updates global indexes asynchronously, so global-index reads can be stale (like replication lag). Best when reads outnumber writes and postings lists aren't too long.

---

Real-world examples and analogies

• Sharding, splitting a giant phone book by region. One book is unmanageable, so you split it: NYC numbers in one volume, LA in another. As long as you know the region (partition key) you grab the right volume instantly; if you don't, you'd have to flip through every volume.
• Hot shard, one checkout lane with the whole crowd. Even with ten lanes, if everyone queues at lane 3 (because that's where this month's data lands), the other nine sit idle. Good sharding is spreading shoppers across lanes.
• Key-range hot spot with timestamps, everyone arriving "now." If you file events by time, all of today's events land in today's drawer, that drawer overflows while last year's drawers gather dust. Prefixing with sensor ID is like filing by department first, then date.
• Mod-N rehashing, renumbering every house when one street is added. Assigning homes by address % N means adding one street forces almost the whole town to get new house numbers. Fixed shards (many "districts") let you just reassign whole districts to a new mail carrier.
• Consistent hashing, adding a slice to a pizza already cut into many. Pre-cut into 1,000 slices and hand some to each guest; a new guest just takes a few slices from others rather than re-cutting the whole pizza.
• Key salting, splitting a celebrity's fan mail. All mail to one star floods one mailbox, so you add a suffix (mailbox_00..99) to spread the deliveries, but now to read all their mail you must check 100 mailboxes.
• Request routing, a switchboard operator. You don't memorize which extension holds which file; you call the operator (routing tier / ZooKeeper) who knows the current map and connects you to the right desk.
• Local vs global index, index in each book vs one master index. Local: every book has its own index, so to find a topic you check the index of every book. Global: one master index sorted by topic, find the topic in one place, though keeping it updated means editing the master whenever any book changes.

---

Cheat-sheet flashcards

Cover the answer and recall it.

• Sharding vs replication? → Split different data across nodes vs copy the same data to nodes.
• How many shards does a record belong to? → Exactly one.
• Primary reason to shard? → Scalability (horizontal scale-out for data/write volume).
• If only reads are the bottleneck? → Use read replicas, not sharding.
• What is a partition key? → The attribute that decides which shard a record goes to.
• What is skew / a hot shard / a hot key? → Uneven load / an overloaded shard / a single high-load key.
• Two main sharding schemes? → Key-range and hash.
• Key-range advantage? → Efficient range scans (sorted keys).
• Key-range disadvantage? → Hot spots when nearby keys are written together.
• Fix for timestamp hot spots? → Prefix the key (e.g., sensor ID) so writes spread.
• How does key-range rebalance? → Split an over-large/over-loaded shard into subranges.
• Hashing's purpose in sharding? → Turn skewed keys into a uniform distribution.
• The problem with hash mod N? → Changing N moves most keys.
• Fixed-shard approach? → Many shards per node; move whole shards on rebalance.
• Limitation of fixed shards? → Can't exceed shard count in nodes; resharding is expensive.
• Hash-range sharding vs key-range? → Range of hash values; adapts in count but loses range-query efficiency.
• Cassandra/ScyllaDB sharding twist? → Random-boundary ranges, many per node, to even out load.
• Two properties of consistent hashing? → Even key spread + minimal key movement when shard count changes.
• "Consistent" in consistent hashing means? → Keys tend to stay put (not replica/ACID consistency).
• Two ways to relieve a hot key? → Dedicated shard, or key salting (random suffix).
• Downside of key salting? → Reads must gather from all salted keys.
• Risk of auto-rebalancing + auto-failure-detection? → Cascading failure (slow node mistaken for dead).
• Three request-routing approaches? → Any node forwards; routing tier; shard-aware client.
• What tracks shard→node assignments? → A coordination service (ZooKeeper/etcd) via consensus.
• Riak's alternative to consensus routing? → Gossip protocol (weaker, split-brain possible).
• Local secondary index (a.k.a.)? → Document-partitioned; each shard indexes its own records.
• Local-index read cost? → Scatter/gather across all shards (tail latency amplification).
• Global secondary index (a.k.a.)? → Term-partitioned; index sharded by indexed value.
• Global-index read vs write cost? → Single-shard reads per term, but multi-shard writes.
• Why are DynamoDB global-index reads sometimes stale? → Indexes updated asynchronously.

---

Common interview questions

1. What is sharding, when do you need it, and how does it relate to replication? (Split data when one node can't hold it / serve writes; combine with replication for fault tolerance.)
2. Compare key-range and hash sharding. (Range scans + hot-spot risk vs even spread + no efficient range queries.)
3. Why is hash(key) % N a bad way to assign shards, and what do you use instead? (Changing N moves most keys; use fixed shards or consistent hashing.)
4. Explain the fixed-number-of-shards approach and its limitations. (Many shards per node, move whole shards; can't exceed shard count, expensive resharding, sizing.)
5. What is consistent hashing and what two properties does it provide? (Even spread + minimal movement; rendezvous/jump variants.)
6. How do you handle a hot key like a celebrity's posts? (Dedicated shard, key salting with read trade-off, adaptive capacity.)
7. What are the risks of automatic rebalancing? (Expensive data movement, overload, cascading failure with false death detection; human-in-the-loop.)
8. Describe the three request-routing approaches and how the routing layer stays current. (Any node / routing tier / shard-aware client; ZooKeeper/etcd via consensus, or gossip.)
9. Why don't secondary indexes map neatly to shards, and what are the two solutions? (Index value isn't the partition key; local vs global.)
10. Local vs global secondary indexes, read/write trade-offs? (Local: cheap writes, scatter/gather reads; global: single-shard reads, multi-shard writes, possibly stale.)
11. How would you shard a multitenant SaaS app, and what are the trade-offs? (Per-tenant shards: isolation/compliance/backup; big tenants, many small tenants, cross-tenant joins.)
12. A write needs to update records in two shards, what's the catch? (Needs a distributed transaction; slower, potential bottleneck, Chapter 8.)

---

Key terms glossary

---

Forward links (where these ideas return)

• Replication of each shard → Chapter 6
• Distributed transactions across shards / global-index writes → Chapter 8
• Network partitions, cascading failures, false death detection → Chapter 9
• Consensus, coordination services, leader election → Chapter 10
• Parallel query execution across shards (analytics) → Chapter 11
• Secondary indexes and postings lists → Chapter 4