Sharding and Partitioning

Partitioning splits one logical dataset into disjoint subsets called partitions. Sharding is partitioning where partitions are placed on different machines; the terms are often used interchangeably. Each record belongs to exactly one partition, and each partition is usually also replicated for fault tolerance.

Partitioning solves two limits of a single node: dataset size beyond one machine's disk or memory, and write or query throughput beyond one machine's CPU and I/O. Reads can be scaled with replicas, but write scaling requires splitting the data.

Partitioning schemes

The partition key is the record attribute used to decide where the record lives. The scheme maps a key to a partition.

Range partitioning

Each partition owns a contiguous range of key values, for example A-F, G-M, N-Z, or timestamp ranges.

• Keys stay sorted within and across partitions, so range scans (WHERE ts BETWEEN a AND b) touch few partitions.
• Partition boundaries can adapt to the data distribution, splitting a range that grows large.
• Monotonically increasing keys (timestamps, auto-increment IDs) send all new writes to the last partition, creating a hot spot.

Used by Bigtable, HBase, and the range-based mode of many distributed SQL databases.

Hash partitioning

A hash function is applied to the key, and the hash decides the partition, either by hash(key) mod N or by assigning hash ranges to partitions.

• Keys spread uniformly, which removes most hot spots caused by skewed or sequential keys.
• Adjacent keys are scattered, so range queries on the key must query every partition.
• With hash(key) mod N, changing N moves almost every key. Real systems avoid this by using a fixed number of partitions or consistent hashing.

Used by Cassandra (with the default partitioner), DynamoDB, and Redis Cluster (16384 hash slots).

Directory-based partitioning

A lookup service stores an explicit mapping from key (or key range) to partition.

• Any placement policy is possible, including moving a single tenant to its own shard.
• The directory is an extra hop and a component that must be highly available and cached correctly.

Comparison

A common compromise is a compound key: hash the first part to choose the partition and sort by the remaining parts inside it. Cassandra's partition key plus clustering columns follow this model. A query for one user's events in a time range hits one partition and reads them in order.

flowchart LR
    C[Client] --> R[Router]
    R -->|"hash(user_id) in slot range 0-5460"| P1[Shard 1]
    R -->|"5461-10922"| P2[Shard 2]
    R -->|"10923-16383"| P3[Shard 3]
    P1 --- R1[Replica 1a]
    P2 --- R2[Replica 2a]
    P3 --- R3[Replica 3a]

Routing requests

A client must find the node that holds a key. Three approaches exist:

1. Any node forwards. The client contacts any node, which forwards the request to the owner (Cassandra coordinators, DynamoDB request routers).
2. Routing tier. A partition-aware proxy holds the mapping and forwards requests (MongoDB mongos, Vitess vtgate).
3. Partition-aware client. The client caches the mapping and connects directly (Redis Cluster clients follow MOVED redirections).

The mapping changes as partitions move. It is kept in a coordination service such as ZooKeeper or etcd, or spread by a gossip protocol.

Hot spots and skew

Even hash partitioning cannot spread load that targets one key, such as a celebrity account or a viral item. All requests for that key hit one partition.

Mitigations:

• Key salting. Append a small random suffix (key#0 to key#9) on writes to spread one hot key across 10 partitions. Reads must query all 10 suffixes and merge results. Apply it only to keys known to be hot, because it complicates every read.
• Caching in front of the store for hot reads (see cache invalidation, stampede and hot keys).
• Splitting the partition when the hot key set is a range rather than a single key.
• Choosing a better partition key so that load correlates less with any single value. Timestamps are a poor leading key for write-heavy data; (device_id, time) is better than time.

Rebalancing and resharding

Rebalancing moves partitions between nodes when nodes are added, removed, or load is uneven. Requirements: load is shared fairly after the change, the database keeps serving during the move, and only the necessary data moves.

Strategies:

Choosing a fixed partition count too low limits future growth, and too high adds per-partition overhead. Changing the partition count later is a resharding operation that rewrites data.

Rebalancing is expensive: it copies data over the network while serving traffic. Many systems require an operator to approve a plan rather than rebalance automatically, since automatic rebalancing combined with automatic failure detection can cascade. A slow node may be marked dead, its partitions moved, and the extra load may then slow other nodes.

Secondary indexes on partitioned data

Queries on an attribute other than the partition key need a secondary index. Two designs exist.

flowchart TB
    subgraph Local["Local index (document-partitioned)"]
        direction LR
        LP1["Partition 1: data + index for its own rows"]
        LP2["Partition 2: data + index for its own rows"]
    end
    subgraph Global["Global index (term-partitioned)"]
        direction LR
        GI1["Index partition: color a-m"]
        GI2["Index partition: color n-z"]
    end
    Q1[Query by color] --> LP1
    Q1 --> LP2
    Q2[Query by color=red] --> GI2

Scatter-gather sends a query to every partition and merges the results. Its latency is determined by the slowest partition, and its cost grows with partition count.

Cross-shard operations

• Joins. Joining tables partitioned on different keys requires moving data between shards. Designs avoid this by choosing the same partition key for related tables (for example tenant_id on all tables) so related rows are colocated, or by denormalizing (see data modeling and denormalization).
• Transactions. An atomic update across partitions needs a distributed commit protocol such as two-phase commit, or a design that keeps each transaction within one partition. Many NoSQL stores support atomicity only within one partition or item.
• Global constraints. Uniqueness on a non-partition-key column requires a global index or a coordination step.
• Global ordering and aggregation. ORDER BY ... LIMIT and counts require querying each partition and merging: each returns its top k, and the coordinator merges.
• Unique IDs. Auto-increment does not work across shards. See design unique ID generator.

Choosing a partition key

Criteria:

1. High cardinality, so there are many more distinct values than partitions.
2. Even access distribution, avoiding one value receiving most traffic.
3. Matches the dominant query, so most queries name the key and hit one partition.
4. Keeps data that is read or transacted together in the same partition.

Multi-tenant systems commonly partition by tenant ID. A very large tenant may then outgrow one partition, so such systems often support moving large tenants to dedicated shards through a directory.

When to shard

Sharding adds routing, rebalancing, cross-shard query, and operational complexity. Prefer these steps first: better indexes, caching, read replicas, vertical scaling, and archiving old data. Shard when the write rate or dataset size exceeds what one primary can handle.

Interview questions

• Why is hash(key) mod N a poor sharding function when N can change? Changing N changes the assigned node for almost all keys, forcing near-total data movement. Fixed partition counts or consistent hashing move only a fraction of keys.
• How do you handle a single hot key on a hash-partitioned store? Hashing cannot split one key. Cache reads, or salt the key across several sub-keys on write and merge on read, applying this only to identified hot keys.
• Local vs global secondary index? A local index is cheap to maintain but queries scatter to all partitions. A global index answers from one index partition but writes touch multiple nodes and the index is often eventually consistent.
• Why is range partitioning bad for time-series ingestion keyed by timestamp? All current writes fall into the newest range, so one node takes all the write load. Prefixing the key with a source ID spreads the writes.