Partitioning Sharding Rebalancing and Hot Keys

Published by Programming Line in System Design and Distributed Systems. Last updated .

Purpose and Outcome

Partitioning is the act of dividing a logical dataset into smaller pieces. Sharding is partitioning where those pieces are placed on different storage nodes, so one table, index, queue, or object namespace can grow beyond the capacity of a single machine. Rebalancing is the controlled movement of partitions when capacity, traffic, or ownership changes. A hot key is a partitioning failure mode: one key or narrow key range receives enough traffic to overload the shard that owns it.

In a system design interview or design review, this topic answers a concrete question: when the data and request rate no longer fit comfortably on one node, how do clients find the right node, how do you add or remove nodes without losing availability, and how do you prevent a popular customer, post, product, or time range from dominating the cluster? By the end of this lesson, you should be able to pick a shard key, explain routing metadata, compare hash and range partitioning, plan a rebalance, and diagnose hot partitions from metrics.

How Partitioning Works Internally

A partitioned storage system separates three ideas that are often conflated. The logical data model is what the application sees, such as users, orders, events, or messages. The partitioning function maps each record or request to a partition identifier. The placement map assigns partitions to physical nodes. Keeping these separate lets you move partitions without changing application identifiers.

With hash partitioning, the system takes a stable key, computes a hash, and maps the hash to a bucket. If user_id is the shard key, all rows for one user can be routed together. The advantage is even distribution when keys are numerous and similarly active. The cost is that range scans across adjacent business values become scatter-gather queries because the hash destroys ordering.

With range partitioning, each partition owns an ordered interval, such as order IDs 0 through 2499 or timestamps for one day. Range partitioning is excellent for ordered reads and archival movement, but it can create hot ranges. A current-day events table, for example, can send nearly every write to the newest range unless the design adds subpartitioning, salting, or time buckets with multiple writers.

Most production systems avoid mapping records directly to physical nodes. They introduce virtual partitions, also called buckets, tablets, vnodes, or shards depending on the database. A cluster might have thousands of virtual partitions but only dozens of storage nodes. The routing layer maps key to virtual partition, then virtual partition to node. Rebalancing moves virtual partitions between nodes, which is more precise than moving arbitrary rows.

Routing Anatomy

A write normally follows this path: validate the request, extract the shard key, compute the partition, locate the current owner in routing metadata, send the write to the owner, and replicate according to the storage system’s consistency policy. A read follows the same routing path if it is key-based. Queries that do not include the shard key either broadcast to many shards or use a secondary index that has its own partitioning strategy.

The critical configuration elements are the shard key, partition count, replica factor, placement policy, routing metadata store, and migration protocol. The shard key determines locality and distribution. Partition count sets movement granularity. Replica factor protects availability. Placement policy spreads data across machines, racks, or zones. Routing metadata tells clients or gateways who owns each partition. The migration protocol decides what happens while ownership is changing.

Example 1: Hash Routing for Users

The smallest practical example is a user service that routes all requests for a user to one of four shards. The code hashes the user ID, converts part of the digest to an integer, and applies modulo arithmetic. The printed routes are deterministic for the listed keys.

import hashlib

SHARDS = ["shard-a", "shard-b", "shard-c", "shard-d"]

def route_user(user_id: str) -> str:
    digest = hashlib.sha256(user_id.encode()).hexdigest()
    bucket = int(digest[:8], 16) % len(SHARDS)
    return SHARDS[bucket]

for user_id in ["u100", "u101", "u102", "u103", "u104"]:
    print(f"{user_id} -> {route_user(user_id)}")

Expected output: u100 routes to shard-d, u101 to shard-c, u102 to shard-a, u103 to shard-c, and u104 to shard-b. This model is simple, but adding a fifth shard changes the modulo result for many keys. That means large data movement and cache churn unless the system uses virtual buckets or consistent hashing.

Example 2: Range Routing for Orders

Range partitioning keeps related ordered values together. The next example sends order IDs into four numeric ranges. A query for orders 2500 through 4999 can go straight to shard-b. A query spanning 2000 through 6000 must contact three shards.

import bisect

# Each tuple is an exclusive upper bound and the shard that owns that range.
RANGES = [(2500, "shard-a"), (5000, "shard-b"), (7500, "shard-c"), (10000, "shard-d")]

def route_order(order_id: int) -> str:
    bounds = [upper for upper, _ in RANGES]
    index = bisect.bisect_right(bounds, order_id)
    return RANGES[index][1]

for order_id in [17, 3400, 7120, 9001]:
    print(f"order {order_id} -> {route_order(order_id)}")

The expected routes are shard-a, shard-b, shard-c, and shard-d respectively. The operational risk appears when IDs are monotonically increasing: new orders all land in the highest active range until it splits. Systems using range partitioning often pre-split future ranges or add a hash suffix so writes spread while reads can still constrain by range.

Example 3: Salting a Hot Tenant

Suppose one tenant has a social feed receiving one thousand writes per second. If tenant_id alone is the shard key, every write for that tenant maps to one virtual bucket. Salting appends a small deterministic suffix, often derived from item ID, message ID, or a random write bucket. Reads then gather from the salt buckets and merge results.

import hashlib
from collections import Counter

def hash_to_virtual_bucket(key: str, buckets: int = 64) -> int:
    return int(hashlib.md5(key.encode()).hexdigest(), 16) % buckets

def salted_feed_key(tenant_id: str, item_id: int, salts: int = 8) -> str:
    return f"{tenant_id}#{item_id % salts}"

plain = Counter(hash_to_virtual_bucket("celebrity-tenant") for _ in range(1000))
salted = Counter(hash_to_virtual_bucket(salted_feed_key("celebrity-tenant", i)) for i in range(1000))

print(f"plain buckets touched: {len(plain)}")
print(f"salted buckets touched: {len(salted)}")
print(f"largest salted bucket load: {max(salted.values())}")

The unsalted key touches one bucket. The salted form touches eight buckets, with the largest salted bucket receiving 125 of the thousand writes in this deterministic example. The trade-off is read complexity: a feed lookup must query eight salted keys, merge by timestamp, and handle partial failures.

Rebalancing Mechanics

Rebalancing is not simply copying files. A typical migration has phases: choose a partition to move, mark the destination as a learner or follower, copy a snapshot, stream changes that occurred during the copy, switch the routing metadata, drain old in-flight requests, and eventually remove the old replica. During this period, the system must define where writes go, whether reads may use the new copy, and how duplicate requests are handled.

Some systems use client-side routing, where application libraries cache the partition map. Others use a gateway or proxy, so clients connect to a stable endpoint and the gateway performs routing. Client-side routing can reduce hop count but makes metadata staleness a practical concern. Gateway routing centralizes control but adds a dependency and a possible bottleneck.

import hashlib

VIRTUAL_BUCKETS = 16
old_map = {bucket: bucket % 4 for bucket in range(VIRTUAL_BUCKETS)}
new_map = {bucket: bucket % 5 for bucket in range(VIRTUAL_BUCKETS)}

def bucket_for_key(key: str) -> int:
    return int(hashlib.sha1(key.encode()).hexdigest(), 16) % VIRTUAL_BUCKETS

keys = [f"user-{i}" for i in range(1000)]
moved = sum(1 for key in keys if old_map[bucket_for_key(key)] != new_map[bucket_for_key(key)])
print(f"keys that change physical shard: {moved}")

This example shows why direct modulo on physical shard count is disruptive. In the sample, 745 of 1000 keys change physical shard when moving from four to five shards. With a larger virtual bucket map, an operator can instead move selected buckets and keep most keys stable.

Design Choices and Trade-offs

Choose a shard key by balancing three forces: write distribution, read locality, and operational flexibility. user_id works well for per-user profiles because reads and writes usually target one user. tenant_id works for small and medium tenants but fails for a tenant whose traffic dwarfs the rest. timestamp helps retention and analytics scans but tends to hot-spot recent writes. composite keys such as tenant_id plus hash(user_id) can preserve tenant-level filtering while spreading heavy tenants.

Fixed partition counts are easy to reason about but can create large partitions as the dataset grows. Many small virtual partitions increase metadata and background task overhead, but they make movement smoother. More replicas improve read availability and failover options, but every write must be propagated to more places. Cross-shard transactions are possible with coordination protocols, but they add latency and failure states, so many large systems design APIs to keep the common transaction inside one shard.

Failure Modes and Troubleshooting

A hot key usually appears as one shard with high CPU, storage queue depth, lock contention, or p99 latency while cluster averages look healthy. Start by grouping request counts and latency by shard, then by partition, then by logical key. If one key dominates, apply request coalescing, caching, salting, or a special-case split for that entity. If many adjacent keys dominate, the issue is a hot range, and range splitting or hash subpartitioning is more appropriate.

A stale routing map shows up as sudden not-owner errors, redirects, or writes accepted by an old owner after a move. Check routing metadata version numbers in clients and servers. Correct by reducing metadata cache TTL, forcing refresh on ownership errors, and making the migration protocol reject stale writes unless they carry a valid epoch.

A rebalance storm occurs when automatic balancing reacts too aggressively. Symptoms include constant partition movement, saturated network links, and read latency rising even though no user workload changed. Diagnose by inspecting balancer decisions, bytes moved per hour, and thresholds for imbalance. Correct by adding movement budgets, cooldown periods, and separate triggers for size imbalance, traffic imbalance, and hardware replacement.

Scatter-gather query overload is another common problem. It appears when a query without the shard key fans out to every shard, often for dashboards or admin search. Verify with query traces that show every shard contacted. Fix by adding a purpose-built index, requiring a narrowing predicate, materializing aggregates, or moving analytical queries to a separate store.

Reliability, Performance, and Security Implications

Partitioning improves horizontal capacity, but it also turns metadata into a reliability dependency. The partition map needs durable storage, versioning, and rollback. Rebalancing consumes disk, network, and cache capacity, so it should be rate-limited and observable at partition granularity. Replication placement should avoid putting all replicas of a hot partition in the same rack or zone.

Shard keys can leak sensitive business structure if exposed directly. Sequential tenant or order ranges may reveal scale or customer concentration. Public APIs should use opaque IDs where possible, validate that a caller is authorized for the routed key, and avoid trusting client-supplied shard hints. A shard hint can be an optimization, but the server must recompute or verify the actual partition.

Hands-on Lab: Model a Safer Shard Plan

Prerequisites: a local Python interpreter and permission to run short scripts. The lab uses only standard library modules. Step 1: list the top three access patterns for a hypothetical order system: get order by ID, list orders for customer, and append payment events. Step 2: select customer_id as the primary shard key for customer order history, then note that order lookup by order_id needs either an order_id-to-customer_id lookup table or a secondary index. Step 3: create 128 virtual buckets and map them onto four physical shards. Step 4: simulate one large customer by assigning that customer 16 salted buckets instead of one. Step 5: plan a rebalance by moving the busiest virtual buckets to a new fifth shard while leaving cold buckets untouched.

Verification: after the simulation, no single physical shard should own a majority of the hot customer’s salted buckets, and ordinary customers should still route deterministically from customer_id to one bucket. Check that each migration has an old owner, new owner, copied byte count, catch-up lag, and metadata version. Cleanup or rollback: keep the old owner read-only until the new owner is caught up, then switch routing. If verification fails, restore the previous partition map version and resume writes on the old owner.

Assessment Exercises

  1. A product catalog is mostly read by product_id, but one flash-sale product receives half of all traffic for ten minutes. Which mitigation would you try first, and what metric proves it worked?
  2. An events table is partitioned by day, and writes to today’s partition are slow while old partitions are idle. Redesign the key without destroying the ability to query one day of events.
  3. Your team wants to add two nodes to a hash-sharded cluster that uses key modulo node count. Explain the data movement risk and propose a virtual bucket migration plan.
  4. A support dashboard runs customer searches without customer_id and times out after contacting every shard. What index or data model change would reduce fan-out?
  5. During rebalancing, clients report not-owner errors for five minutes after the control plane changed the map. What client and server behavior would you inspect?

Summary

Partitioning and sharding scale storage by routing each key to a smaller ownership unit. The hard parts are choosing a shard key that matches access patterns, keeping routing metadata correct, moving partitions gradually, and recognizing hot keys before they become cluster-wide incidents. Hashing spreads load, ranges preserve order, virtual buckets make rebalancing manageable, and salting splits exceptional keys at the cost of more complex reads. Good system design makes those trade-offs explicit instead of treating sharding as a capacity checkbox.