11. Partitioning & Sharding

When one machine can't hold all your data, you split it across many. Partitioning (= sharding) is how. The decision is irreversible-ish — pick wrong and you'll spend years regretting it.

11.1 Why partition

Three reasons:

1. Storage: data doesn't fit on one node.
2. Throughput: writes/reads exceed one node's capacity.
3. Locality: serve users from data near them (geo-partitioning).

Note: replication and partitioning are orthogonal. A typical large system has N partitions × R replicas = N×R nodes.

11.2 Partitioning strategies

Range partitioning

Each partition owns a contiguous range of keys. Rows sorted by key.

Examples: Bigtable, HBase, Spanner, distributed SQL DBs.

Pros: range scans are efficient (single partition or a few adjacent). Cons: hot partitions if keys correlate with time/load. Time-prefixed keys (timestamps as PK) write to one partition exclusively → write hotspot.

Mitigation: prefix keys with a hash, or shard high-write entities differently.

Hash partitioning

Hash the key, mod by partition count.

Pros: uniform load distribution; trivial to implement. Cons: range queries impossible (consecutive keys land in different partitions).

Examples: Cassandra (with token range), DynamoDB, Memcached, Redis Cluster.

Hybrid: hashed prefix + range suffix

Cassandra: partition key (hash) + clustering key (sorted within partition). Get uniform distribution across nodes, range queries within a partition.

Bigtable: row key is a string; you choose to make it a hash prefix + meaningful suffix. Common pattern: (reverse(domain)/page_id) for web pages.

Geographic partitioning

Partition by region (us-east, eu-west). User's data lives near user. Cross-region writes are rare/forbidden. Spanner, Cosmos DB support this.

Lookup-table / directory-based

A separate service maps keys → partitions. Flexible (rebalance freely) at the cost of an extra hop and a single point of failure.

11.3 Hot keys / skew

Even with good hashing, some keys are inherently hot (a celebrity's profile, a popular product).

Mitigations:

• Replicate the hot key across partitions. Reads spread; writes are still on one. Update propagation needed.
• Application-level fanning: split the hot entity into sub-entities (celebrity:1#bucket:0..99); aggregate on read.
• Caching: hot keys cache 99% of reads.
• Salting: append a random suffix to the hot key during writes; aggregate on read. Used for time-series counters.

11.4 Partitioning secondary indexes

Two strategies, each painful in its own way.

Local secondary indexes (per partition)

Each partition indexes its own data. Reads must scatter-gather across all partitions, then merge.

Pros: writes are simple (only the local index is updated). Cons: reads are slow (N partitions queried).

Used by: Cassandra (default), Riak, MongoDB.

Global secondary indexes

A separate index, partitioned by the indexed column. Reads hit one partition.

Pros: fast reads. Cons: writes are distributed transactions across data partition + index partition. Often eventually consistent (DynamoDB GSIs).

11.5 Rebalancing

When a node is added/removed, partitions must move.

Bad: hash mod N

If N changes, every key remaps. Catastrophic.

Fixed number of partitions

Pick a large number up front (e.g., 1024 partitions). Each node holds many partitions. Adding a node: move some partitions to it. Number of partitions doesn't change.

Used by: Riak, Elasticsearch, Cassandra (with vnodes), etcd.

Drawback: partition size grows as data grows; you can't easily split.

Dynamic partitioning

Partition splits when it gets too big; merges when too small. Range partitions specifically.

Used by: HBase, Bigtable, Spanner, MongoDB.

Drawback: splitting is operational work; rebalancing during bursts can interfere with traffic.

Consistent hashing

(See chapter 17.) When adding/removing nodes, only ~K/N keys move.

Used by: Memcached, Cassandra, Riak, Dynamo.

Rebalancing is dangerous

Moving partitions = lots of data. Throttle. Schedule during low traffic. Beware: a botched rebalance can cascade — Bigtable, Cassandra, all have horror stories.

Most production systems make rebalancing operator-initiated, not automatic. Automatic rebalancing during a slow query storm could pile work on top of work.

11.6 Routing

Client needs to find the right partition. Three approaches:

1. Client-side: client knows partition map; sends to right node. Used by some KV clusters (Redis Cluster client, Memcached clients).
2. Routing tier: a stateless proxy looks up partition and forwards (mongos, CockroachDB SQL gateway). Adds a hop.
3. Any node accepts, redirects: if not local, node forwards to right node. Cassandra works this way.

Coordination service (ZooKeeper, etcd, Chubby) often holds the partition map.

11.7 Choosing the partition key

This is the most consequential decision in any sharded system. Get it wrong, and rebalancing is brutal.

Criteria:

• Even distribution: avoid hot partitions.
• Affinity with access patterns: queries should be answerable by one or few partitions.
• Stable: doesn't change over the life of the entity.

For social: user_id. For multi-tenant SaaS: tenant_id. For time-series: device_id or (device_id, time_bucket).

Anti-patterns:

• Sharding by timestamp alone (writes pile on most-recent partition).
• Sharding by monotonically increasing ID (same problem).
• Sharding by status or type (skew, since some statuses dominate).

11.8 Cross-partition operations

The hardest part of sharded systems.

Cross-partition queries

Scatter → gather. Send to all (or relevant) partitions, merge results in the coordinator. Costs grow with shard count and tail latency.

Cross-partition transactions

Need a distributed transaction protocol (2PC, Spanner-style with TrueTime, or saga). Slow and harder to operate.

Joins across partitions

Generally avoided. Strategies:

• Co-locate related data in the same partition (Cassandra: clustering key; Cosmos DB: partition key).
• Denormalize: write the joined data together.
• Stream-table joins in stream processing (Flink, Kafka Streams).
• Fan-out at write: precompute the joined result and store it.

11.9 Multi-tenant sharding patterns

For SaaS:

1. Shared tables, tenant_id column: one DB, every row tagged. Cheap. Risk: noisy neighbor; bug leaks data across tenants. Use Postgres Row-Level Security to enforce.
2. Shared DB, schema per tenant: each tenant has own schema. Clean isolation; Postgres scales to ~1000s of schemas before the catalog hurts.
3. Database per tenant: maximum isolation. High operational overhead. Common for enterprise plans.
4. Shard by tenant: tenants distributed across shards; many tenants per shard. Best of both.

Big tenants (whales) often get dedicated shards.

11.10 Practical example: Twitter shards by user_id

• Tweets: shard by user_id of author. A user's tweets all live on one shard.
• Timelines: shard by user_id of viewer. Pre-computed per user (fan-out on write).
• Direct messages: shard by user_id of either party (depends on access pattern).

Hot shard problem: a celebrity's tweet shard sees write spikes. Mitigation: replicate the celebrity write through a cache; serve their followers from cache.

11.11 Practical example: Uber shards by trip_id

Uber's old system sharded riders, drivers, trips by user. Cross-shard joins (e.g., "find trip + payment + receipt") were painful. Newer architecture: separate services with their own sharding, joined via APIs or events.

11.12 Sharding signals

Signs your single-DB is past its breaking point:

• p99 query latency rising
• Vacuum can't keep up (Postgres bloat)
• WAL replication lag growing
• Write IOPS at hardware limit
• Connection pool exhaustion despite pooling
• Largest tables > 1 TB

Before sharding, consider:

• Read replicas for read scale.
• Caching for read load.
• Vertical scale (RDS r6i.32xl is huge).
• Splitting domains into separate services (and DBs).

Key takeaways

• Partitioning unlocks horizontal scale at the cost of cross-partition operations.
• Hash for uniform distribution; range for ordered access; geo for locality.
• Choose the partition key carefully — it's hard to change.
• Hot keys / skew are inevitable; have a strategy.
• Rebalancing is dangerous; operator-driven, throttled.
• Vertical scale and caching first; shard only when you must.