Database Sharding Strategies: From Theory to Production

Sharding is the practice of distributing a database across multiple physical nodes, each holding a subset of the data. It's the correct solution for exactly one problem: a single database instance cannot handle your read/write throughput or storage requirements. It is not a general-purpose scalability strategy, and the operational costs are substantial. Start sharding only when you've exhausted vertical scaling, read replicas, caching, and query optimization.

The Sharding Key: The Most Important Decision

Your sharding key determines which shard stores each row and which shard must be queried to retrieve it. Getting this wrong means either hot spots (all traffic hitting one shard) or queries that must fan out across all shards to get an answer.

Properties of a Good Sharding Key

High cardinality. A key with few distinct values (e.g., a boolean, an enum with 5 values) can never distribute data evenly. You need at least as many distinct values as you plan to have shards, ideally orders of magnitude more.

Even distribution. The values should be roughly uniformly distributed. User IDs work well because users don't cluster. Geographic region works poorly if 70% of your users are in one country.

Query locality. The most frequent queries should be answerable from a single shard. If you're building a multi-tenant SaaS application, tenant_id is usually the right sharding key because all data for a tenant lives on one shard.

Immutability. Changing a row's sharding key requires moving it to a different shard — an expensive operation. The key must be stable for the lifetime of the row.

For most web applications, user_id or tenant_id satisfies all four properties.

Range Sharding vs. Hash Sharding

Range Sharding

Each shard holds a contiguous range of key values. Shard 1 holds keys 0-999999, Shard 2 holds 1000000-1999999, etc.

-- PostgreSQL range partitioning (a form of range sharding)
CREATE TABLE orders (
    id BIGINT,
    tenant_id BIGINT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL,
    total_cents INTEGER NOT NULL
) PARTITION BY RANGE (tenant_id);

CREATE TABLE orders_shard_1 PARTITION OF orders
    FOR VALUES FROM (1) TO (100001);

CREATE TABLE orders_shard_2 PARTITION OF orders
    FOR VALUES FROM (100001) TO (200001);

Advantages: range queries on the shard key are efficient (a single shard or a small number of contiguous shards). Disadvantages: sequential key insertion (e.g., auto-increment IDs, time-series data) causes all writes to hit the newest shard while older shards sit idle — a classic hot spot.

Hash Sharding

Apply a hash function to the key to determine the shard:

def get_shard(tenant_id: int, num_shards: int) -> int:
    return hash(str(tenant_id)) % num_shards

# Consistent hashing implementation to minimize rebalancing
import hashlib

class ConsistentHashRing:
    def __init__(self, shards: list[str], replicas: int = 150):
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
        for shard in shards:
            for i in range(replicas):
                key = self._hash(f"{shard}-{i}")
                self.ring[key] = shard
        self.sorted_keys = sorted(self.ring.keys())
    
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_shard(self, key: str) -> str:
        hash_val = self._hash(key)
        for ring_key in self.sorted_keys:
            if hash_val <= ring_key:
                return self.ring[ring_key]
        return self.ring[self.sorted_keys[0]]

Consistent hashing is essential when you plan to add shards. With naive modulo hashing (key % num_shards), adding one shard remaps approximately n/(n+1) of all keys — a massive data migration. Consistent hashing remaps only 1/(n+1) of keys, making shard addition practical.

Cross-Shard Queries

The most painful consequence of sharding: queries that need data from multiple shards must be executed on each relevant shard and the results merged in application code or a query coordinator.

Fan-Out Queries

async def get_global_revenue_by_product(
    start_date: date,
    end_date: date,
    shards: list[DatabaseShard]
) -> dict[int, int]:
    # Execute on all shards in parallel
    shard_results = await asyncio.gather(*[
        shard.execute("""
            SELECT product_id, SUM(total_cents) as revenue
            FROM orders
            WHERE created_at BETWEEN %s AND %s
            GROUP BY product_id
        """, [start_date, end_date])
        for shard in shards
    ])
    
    # Merge results in application
    global_revenue: dict[int, int] = defaultdict(int)
    for shard_result in shard_results:
        for row in shard_result:
            global_revenue[row['product_id']] += row['revenue']
    
    return dict(global_revenue)

Fan-out queries scale poorly — each query touches every shard regardless of data distribution, and total latency is bounded by the slowest shard. For analytics and reporting, materialize cross-shard aggregations into a separate analytical store (a data warehouse or read replica that aggregates across shards) on a schedule.

Scatter-Gather with Early Termination

For queries like "find any 10 recent orders across all tenants," use scatter-gather with a limit and early termination:

async def find_recent_orders_global(limit: int = 10) -> list[Order]:
    # Ask each shard for more than we need, take the globally most recent
    shard_limit = limit * 2  # get extra to handle merging
    
    shard_results = await asyncio.gather(*[
        shard.execute(
            "SELECT * FROM orders ORDER BY created_at DESC LIMIT %s",
            [shard_limit]
        )
        for shard in shards
    ])
    
    all_orders = [order for result in shard_results for order in result]
    all_orders.sort(key=lambda o: o.created_at, reverse=True)
    return all_orders[:limit]

Rebalancing Shards

As data grows unevenly — some tenants are 100x larger than others — you'll need to move tenants between shards (rebalancing).

Double-Write Migration Pattern

Never copy data with the source shard offline. Instead:

1. Enable double-write. The application writes to both the source shard and the destination shard for the tenant being migrated.
2. Backfill historical data. Copy historical rows from source to destination. Track progress.
3. Verify consistency. Compare row counts and checksums between source and destination.
4. Switch reads. Update the routing table to read from the destination shard. Source continues receiving double-writes.
5. Stop double-write. Update routing to write only to destination. Source is now stale.
6. Delete from source. After a safety window, drop the tenant's data from the source shard.

class ShardMigration:
    async def migrate_tenant(
        self,
        tenant_id: int,
        source_shard: DatabaseShard,
        dest_shard: DatabaseShard,
        routing_table: RoutingTable
    ) -> None:
        # Step 1: Enable double-write at the router level
        await routing_table.set_double_write(tenant_id, source_shard, dest_shard)
        
        # Step 2: Backfill — copy in chunks to avoid lock contention
        last_id = 0
        while True:
            rows = await source_shard.execute(
                "SELECT * FROM orders WHERE tenant_id = %s AND id > %s ORDER BY id LIMIT 1000",
                [tenant_id, last_id]
            )
            if not rows:
                break
            await dest_shard.bulk_insert('orders', rows)
            last_id = rows[-1]['id']
        
        # Step 3: Verify
        source_count = await source_shard.scalar(
            "SELECT COUNT(*) FROM orders WHERE tenant_id = %s", [tenant_id])
        dest_count = await dest_shard.scalar(
            "SELECT COUNT(*) FROM orders WHERE tenant_id = %s", [tenant_id])
        assert source_count == dest_count, f"Count mismatch: {source_count} vs {dest_count}"
        
        # Step 4 & 5: Switch reads and writes atomically
        await routing_table.set_primary_shard(tenant_id, dest_shard)
        
        # Step 6: Cleanup after safety window (run as separate job)
        await self.schedule_cleanup(tenant_id, source_shard, delay=timedelta(hours=24))

Operational Reality

Sharding introduces complexity that compounds across your entire engineering operation:

• Schema migrations must be applied to all shards in sequence (or in parallel with careful ordering)
• Backup and restore must operate per-shard, with cross-shard consistency guaranteed at the application level
• Monitoring must aggregate metrics across shards — per-shard and fleet-wide dashboards both matter
• Developer environments need a sharded test setup or a single-shard emulation mode

Before sharding, seriously evaluate Citus (PostgreSQL extension for transparent sharding), PlanetScale (MySQL-based), or Google Spanner (managed globally distributed SQL). These managed solutions take on most of the operational burden, at the cost of less control over placement and migration behavior.

The questions "do we need to shard?" and "when do we shard?" are as important as "how do we shard?" The answer to the first is almost always "not yet." Exhaust your alternatives first. When you do shard, choose the sharding key with extreme care — it's the hardest thing to change after the fact.