Database Sharding Explained: Scale to Millions of Users

Database sharding is the practice of splitting a single database into multiple smaller databases (shards), each holding a subset of the data, so that write load and storage are distributed across many machines. You need it when a single database server can no longer handle the write throughput or storage volume your application demands — typically after you've exhausted indexing, caching, read replicas, and vertical scaling.

Your user table has 50 million rows. Queries are slow. You've added indexes, upgraded to a bigger server, added read replicas. But writes are still the bottleneck.

This is the problem sharding solves. Instead of one big database, you split the data across multiple databases — each one handles a subset of the data. Every write goes to one shard. Every read (for data you can identify by shard key) hits one shard. Total write capacity scales linearly with the number of shards.

What Is Sharding?

Sharding is horizontal partitioning — splitting a table's rows across multiple database instances (called shards).

Each shard has the same schema but different data. Together, they hold the complete dataset.

Without sharding:
Database → Users table (50M rows)
 
With sharding:
Shard 1 → Users 1–12.5M
Shard 2 → Users 12.5M–25M
Shard 3 → Users 25M–37.5M
Shard 4 → Users 37.5M–50M

Each shard can be on a different server. Each server handles 25% of the write load.

Sharding vs partitioning: Partitioning splits data within a single database instance (e.g., PostgreSQL table partitioning with pg_partman). Sharding splits data across multiple database instances — separate servers, separate connections. Partitioning helps with query planning and maintenance; sharding helps with write throughput and storage scale. You can use both at the same time.

Sharding Strategies Compared

The strategy you pick determines your data distribution, query patterns, and how painful resharding will be. There is no universally correct answer — the right choice depends on your access patterns.

Range-Based Sharding

Split data based on ranges of the shard key. User IDs 1–1M go to Shard 1, 1M–2M to Shard 2, etc.

def get_shard(user_id: int, num_shards: int = 4) -> str:
    shard_size = 1_000_000
    shard_num = user_id // shard_size
    return f"db_shard_{shard_num % num_shards}"

Advantage: Simple. Range queries (find all users created between date X and Y) are efficient — they usually hit one or two shards.

Problem: Hot spots. If most active users have high IDs (recently created), Shard 4 handles all the load while Shard 1 sits idle. Range sharding on created_at has the same problem — your most recent shard takes all new writes.

Hash-Based Sharding

Hash the shard key and use modulo to pick a shard.

import hashlib
 
def get_shard(user_id: int, num_shards: int = 4) -> str:
    hash_value = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
    shard_num = hash_value % num_shards
    return f"db_shard_{shard_num}"

Advantage: Even distribution. No hot spots.

Problem: Range queries become expensive — you have to check all shards. Resharding (adding more shards) is painful — you have to move data around.

Solution to resharding: Use consistent hashing instead of simple modulo. Cassandra and DynamoDB do this.

Directory-Based Sharding

A lookup service (routing table) maps each key to its shard.

# Lookup table stored in Redis or a database
routing_table = {
    "user:1": "shard_1",
    "user:2": "shard_3",
}
 
def get_shard(user_id: int) -> str:
    return redis.get(f"routing:user:{user_id}") or "shard_default"

Advantage: Most flexible. You can move data between shards easily by updating the routing table.

Problem: The routing service becomes a bottleneck and a single point of failure. Extra network hop per query.

Geographic Sharding

Route traffic based on the user's geographic region. European users go to EU shards, US users to US shards.

REGION_SHARDS = {
    "EU": "postgres-eu-west.internal",
    "US": "postgres-us-east.internal",
    "APAC": "postgres-ap-southeast.internal",
}
 
def get_shard(user_region: str) -> str:
    return REGION_SHARDS.get(user_region, REGION_SHARDS["US"])

Advantage: Data residency compliance (GDPR keeps EU data in EU), reduced read latency for users.

Problem: Growth is uneven. If your US user base grows 10× and EU stays flat, your US shard becomes a bottleneck while EU idles. You need a strategy to sub-shard within regions.

Choosing a Shard Key

The shard key determines how data is distributed. A bad shard key creates hot spots and limits which queries you can serve efficiently. This decision is the hardest part of sharding — changing it later requires moving all your data.

Good shard keys:

• High cardinality (many unique values)
• Evenly distributed access patterns
• Rarely changes (changing a shard key means moving data)
• Matches your most frequent query pattern

Common choices:

• user_id — if most queries are per-user
• tenant_id — for multi-tenant SaaS apps
• created_at (with hash) — for time-series data
• Geographic region — if users are clustered geographically

Avoid:

• Boolean fields — only 2 shards max
• Low-cardinality fields — uneven distribution
• Fields that change frequently
• Fields your queries never filter on

Cross-Shard Queries: The Hard Problem

Single-shard queries — where the query includes the shard key and hits exactly one shard — are fast and cheap. Cross-shard queries are where sharding gets painful.

Scatter-Gather Queries

When a query cannot be routed to a single shard, you must scatter the query to all shards and gather the results:

import asyncio
from typing import List, Dict
 
async def query_all_shards(query: str, params: dict) -> List[Dict]:
    """Scatter-gather: run query on all shards, merge results."""
    tasks = [
        run_query_on_shard(shard, query, params)
        for shard in ALL_SHARDS
    ]
    results_per_shard = await asyncio.gather(*tasks)
    # Flatten and sort merged results
    merged = [row for shard_rows in results_per_shard for row in shard_rows]
    return sorted(merged, key=lambda r: r["created_at"], reverse=True)

This works, but it has serious costs:

• Latency: Your total latency is now the latency of your slowest shard
• Tail latency amplification: With 16 shards, p99 latency of a single shard becomes the p50 of your scatter-gather
• Resource usage: One logical query becomes N physical queries

For COUNT(*), SUM(), or AVG() across all users, you're running N aggregation queries and summing the results in application code. For LIMIT 10 ORDER BY score DESC across all users, you must fetch the top 10 from each shard and merge-sort them — the correct top-10 requires fetching 10 * N rows in the worst case.

Why Shard Keys Matter for Query Routing

The shard key is the routing key. Every query that includes the shard key as a filter can be routed to exactly one shard. Every query that does not include the shard key must scatter to all shards.

This means your data model and query patterns must be designed around the shard key:

-- Sharded on user_id
 
-- Single-shard: efficient
SELECT * FROM orders WHERE user_id = 12345;
 
-- Cross-shard: scatter-gather to all shards
SELECT * FROM orders WHERE product_id = 999;
 
-- Cross-shard: cannot be served from one shard
SELECT COUNT(*) FROM orders WHERE created_at > '2026-01-01';

Avoiding Cross-Shard Joins

SQL JOINs across shards are impossible at the database level. You have two options:

Option 1: Denormalize data onto the shard. If you shard orders by user_id and need the user's email alongside each order, store the email directly in the orders table. Yes, this is duplication. It's intentional.

Option 2: Application-level join. Fetch from Shard A, collect IDs, query Shard B by those IDs, join in application memory. This works but adds round trips and complexity.

The practical rule: design your schema so that queries your application runs on the hot path can be answered from a single shard. Accept scatter-gather for analytics, reporting, and background jobs where latency tolerance is higher.

Resharding: When You Get It Wrong

You sharded with 4 shards. Now you have 8× the traffic and need 32 shards. Or you picked the wrong shard key and one shard is taking 80% of the load. Either way, you need to reshard.

Resharding is the most operationally painful part of sharding. Done wrong, it causes downtime and data loss.

What Resharding Involves

With simple modulo-based hash sharding (hash(key) % N), adding a new shard invalidates the routing for almost every key. If N goes from 4 to 5, key % 4 != key % 5 for most keys. You must move the majority of your data.

Adding Shard 5 with modulo hashing:
- key 100: hash=400, 400 % 4 = 0 → Shard 0
            hash=400, 400 % 5 = 0 → Shard 0 (no move needed)
- key 101: hash=404, 404 % 4 = 0 → Shard 0
            hash=404, 404 % 5 = 4 → Shard 4 (must move)
- Typically ~80% of keys must move when adding one shard

Consistent Hashing for Resharding

Consistent hashing is the standard solution. Rather than hash(key) % N, you map both keys and shards to positions on a ring. A key is served by the nearest shard clockwise on the ring.

When you add a new shard, it takes over a portion of one existing shard's key range. Only the keys in that range need to move — typically 1/N of total data.

import bisect
import hashlib
 
class ConsistentHashRing:
    def __init__(self, replicas: int = 150):
        self.replicas = replicas
        self.ring: dict[int, str] = {}
        self.sorted_keys: list[int] = []
 
    def add_shard(self, shard_name: str):
        for i in range(self.replicas):
            key = self._hash(f"{shard_name}:{i}")
            self.ring[key] = shard_name
            bisect.insort(self.sorted_keys, key)
 
    def get_shard(self, key: str) -> str:
        if not self.ring:
            raise ValueError("No shards available")
        hash_val = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, hash_val) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]
 
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

When add_shard("shard_5") is called, only the keys that hash into the new shard's arc need to be migrated. All other keys continue to map to their current shards.

Blue/Green Resharding

For production resharding without downtime:

1. Stand up new shard topology (blue = old, green = new)
2. Dual-write: Write all new data to both old and new shards
3. Backfill: Copy existing data from old shards to new shards, chunked to avoid overwhelming the source
4. Verify: Compare checksums or row counts between old and new
5. Cutover: Switch reads to green; stop writing to blue
6. Drain: Keep blue alive for a rollback window (24–48 hours), then decommission

class DualWriteShardRouter:
    def __init__(self, old_router, new_router, write_to_new: bool = True):
        self.old = old_router
        self.new = new_router
        self.write_to_new = write_to_new
 
    def write(self, key: str, data: dict):
        # Always write to old (source of truth during migration)
        self.old.get_shard(key).insert(data)
        # Async write to new during migration window
        if self.write_to_new:
            self.new.get_shard(key).insert_async(data)
 
    def read(self, key: str, read_from_new: bool = False):
        router = self.new if read_from_new else self.old
        return router.get_shard(key).query(key)

The dual-write window is the critical period. Any failure to write to the new shard during this window creates a gap you must reconcile before cutover.

Sharding with PostgreSQL

PostgreSQL doesn't shard natively, but there are several approaches depending on your scale and tolerance for complexity.

Citus Extension

Citus is the most mature PostgreSQL sharding solution. It adds a coordinator node that routes queries to worker nodes (shards), and it handles distributed query planning — including some cross-shard aggregations.

-- On the Citus coordinator
SELECT create_distributed_table('orders', 'user_id');
 
-- Citus routes this to the correct shard automatically
SELECT * FROM orders WHERE user_id = 12345;
 
-- Citus handles this scatter-gather internally
SELECT COUNT(*), SUM(amount) FROM orders WHERE created_at > '2026-01-01';

Citus is production-ready — it powers large deployments at Microsoft (Azure Cosmos DB for PostgreSQL) and has been battle-tested at companies like Heap Analytics. It supports distributed JOINs when joining on the distribution key, and it handles resharding via rebalance_table_shards().

Logical Sharding with pg_partman

If you want sharding within a single PostgreSQL instance (not across servers), pg_partman provides declarative table partitioning managed automatically:

-- Create parent table
CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT NOT NULL,
    amount NUMERIC,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
) PARTITION BY RANGE (created_at);
 
-- pg_partman manages partition creation and maintenance
SELECT partman.create_parent(
    p_parent_table => 'public.orders',
    p_control => 'created_at',
    p_interval => '1 month',
    p_premake => 3
);

This is partitioning, not sharding — all partitions live on one server. But it enables partition pruning (PostgreSQL only scans relevant partitions), parallel query across partitions, and easier data lifecycle management (drop old partitions instead of DELETE).

Application-Level Sharding

The simplest approach for many teams: manage shard routing entirely in application code, connect to multiple PostgreSQL instances.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
 
SHARD_CONNECTIONS = {
    0: create_engine("postgresql://user:pass@shard-0.db.internal/app"),
    1: create_engine("postgresql://user:pass@shard-1.db.internal/app"),
    2: create_engine("postgresql://user:pass@shard-2.db.internal/app"),
    3: create_engine("postgresql://user:pass@shard-3.db.internal/app"),
}
 
def get_session(user_id: int):
    shard_num = user_id % len(SHARD_CONNECTIONS)
    engine = SHARD_CONNECTIONS[shard_num]
    Session = sessionmaker(bind=engine)
    return Session()
 
# Usage
with get_session(user_id=12345) as session:
    orders = session.query(Order).filter_by(user_id=12345).all()

This approach gives you full control, no extension dependencies, and works with any ORM. The downside is that all routing logic lives in your application — schema migrations must be run against each shard, and cross-shard queries require explicit scatter-gather in your code.

Alternatives to Sharding

Sharding is expensive to implement, maintain, and operate. Before committing to it, exhaust these options:

Read replicas: Most applications are read-heavy. Adding PostgreSQL streaming replicas scales read capacity horizontally without sharding complexity. If your write load is modest and reads are the bottleneck, replicas are 10× simpler.

Caching: A well-placed Redis cache in front of your database can absorb 80–95% of read load. Many databases that appear to need sharding are actually cache-starved.

Vertical scaling: Modern cloud instances are large. A db.r6g.16xlarge on RDS gives you 512 GB RAM and 64 vCPUs. Vertical scaling has a ceiling, but that ceiling is higher than most teams reach. The cost is usually lower than the engineering time to implement sharding.

Connection pooling: PgBouncer in transaction pooling mode dramatically reduces connection overhead. Databases that appear write-bottlenecked sometimes recover entirely after adding a connection pool.

Table archival: Move old data to cold storage or a separate archive database. If 80% of your rows are from users inactive for 2+ years, archiving them reduces your hot dataset by 80%.

Partial indexes and better query plans: Slow writes sometimes trace back to index maintenance overhead, not table size. EXPLAIN ANALYZE before you consider sharding.

When NOT to shard: If your dataset fits on a single server with headroom, if your write throughput is under ~10k writes/second, if you have fewer than ~100M rows in your largest table, or if you're a team of fewer than 10 engineers — do not shard. The operational complexity is disproportionate to the benefit.

Sharding Anti-Patterns

Sharding Too Early

The most common mistake. Teams shard because they're nervous about future scale, not because they've measured a current bottleneck. Premature sharding adds years of complexity before it adds any value.

Rule of thumb: shard when you've measured that a specific bottleneck (write throughput, storage, or connection limits) cannot be addressed by the alternatives above. Not before.

Wrong Shard Key Choice

Choosing a shard key that doesn't match your access patterns forces cross-shard queries on your hot path. Choosing a shard key with low cardinality (e.g., status with 3 values) limits you to 3 shards maximum. Choosing a monotonically increasing key (e.g., id with simple range sharding) concentrates all new writes on one shard.

Signs you chose the wrong shard key:

• One shard consistently at high CPU/IOPS while others idle (hotspot)
• Most of your queries do scatter-gather even though you expected single-shard reads
• You find yourself constantly querying by a field that isn't the shard key

Hotspot Shards

A hotspot is a shard receiving disproportionate traffic. Common causes:

• Celebrity problem: If you shard social media posts by user_id and a celebrity with 50M followers posts, all their data hits one shard — and all their followers' read requests hit that shard.
• Temporal skew: Range sharding on time means all new writes go to the "latest" shard.
• Uneven key distribution: Some shard key values are more popular than others (e.g., sharding e-commerce orders by category_id and electronics has 10× the volume of garden tools).

Mitigations: add a random suffix to the shard key for high-traffic entities (write fan-out), use consistent hashing, or add a small number of "overflow" shards specifically for hot keys.

Key Takeaways

• Sharding splits data across multiple databases to scale write throughput and storage
• Hash-based sharding distributes evenly; range-based is better for range queries; consistent hashing reduces migration cost when resharding
• The shard key is the most critical decision — it's hard to change, and it determines which queries are cheap vs expensive
• Cross-shard queries (scatter-gather) are the primary operational cost of sharding — design your schema so your hot-path queries hit one shard
• Resharding requires careful migration: consistent hashing + blue/green deployment
• PostgreSQL sharding options: Citus (mature, distributed), pg_partman (single-node partitioning), or application-level routing
• Try indexing, caching, read replicas, vertical scaling, and archival before sharding
• Shard too early and you pay years of complexity tax for no benefit

Sharding is powerful but complex. Use it when you have to, not because you can.

FAQ

What is database sharding and why do you need it?

Database sharding is the process of splitting a single database into multiple smaller databases (shards), each holding a subset of the data. You need it when a single database server can no longer handle your write throughput or storage volume — typically after exhausting indexing optimizations, read replicas, caching, and vertical scaling. Sharding is a last resort: it solves write scale at the cost of significant operational complexity.

What is a shard key and how do I choose one?

The shard key is the field used to route each row to its shard. Every write and read goes to the shard determined by hashing or ranging on this key. Choose a shard key with high cardinality (many unique values), even distribution across values, stability (it rarely changes), and alignment with your most frequent query filter. user_id or tenant_id are common choices because most application queries filter by them. Avoid boolean fields, low-cardinality fields, and fields that change frequently.

What is the difference between sharding and partitioning?

Partitioning splits a large table into smaller pieces within a single database instance. Sharding splits data across multiple database instances (separate servers). Partitioning helps with query planning, parallel scans, and data lifecycle management. Sharding helps with write throughput and storage scale across machines. You can use both: partition tables within each shard for query performance, while sharding across servers for write scale.

How do cross-shard queries work?

When a query cannot be routed to a single shard (because it doesn't include the shard key), it must scatter to all shards and gather results. Each shard executes the query independently, returns its results, and your application (or a coordinator like Citus) merges them. This is called scatter-gather. It works, but it multiplies latency and resource usage by the number of shards. The goal of shard key selection is to ensure your hot-path queries always hit a single shard.

When should I shard my database?

Shard when: (1) you've measured that write throughput is the bottleneck and vertical scaling is no longer cost-effective, (2) your dataset is larger than fits on a single server, or (3) you need geographic data placement for compliance. Do not shard because you're afraid of future scale, because your dataset is under ~100M rows, or because reads (not writes) are slow — reads are better addressed by replicas and caching.

What are the alternatives to database sharding?

In rough order of simplicity: add indexes and query optimizations, add a connection pool (PgBouncer), add caching (Redis), add read replicas, archive old data to cold storage, scale vertically to a larger instance, and partition large tables with pg_partman. Sharding should come after all of these have been genuinely exhausted, not as a first response to performance concerns.

How do companies like Instagram shard their databases?

Instagram sharded user media by user_id. All photos for a given user live on one shard, so profile views — the hottest read path — never require cross-shard queries. For cross-cutting queries like hashtag search, Instagram built a separate search index rather than doing scatter-gather across shards. The architecture insight: shard for your highest-traffic hot path, and build purpose-built secondary indexes or data stores for queries that don't fit the shard key. This pattern — shard for the common case, specialized systems for the edge cases — is the template most large-scale systems follow.

Related reading: Database Indexing Explained · CAP Theorem Explained · Consistent Hashing