Database Sharding Strategies — Horizontal Partitioning, Consistent Hashing, Range Sharding

In this tutorial, you'll learn about Database Sharding Strategies. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Database Sharding is a horizontal scaling technique that splits a large database into smaller, independent shards distributed across multiple servers, enabling linear performance growth for high-throughput applications.

What You'll Learn

You'll understand range sharding, consistent hashing, directory-based sharding, and geo-sharding; how to choose a shard key, handle resharding, manage cross-shard queries, and avoid common pitfalls like hotspot shards and join failures.

Why It Matters

A single database server can handle only so many queries per second. Beyond that, you need to scale horizontally. Doda Browser processes search queries from millions of users; without sharding, the entire search index would be bottlenecked by a single database server.

Real-World Use

A social media platform with 200 million users stored all posts in one database. Queries became slower every week. By sharding on user_id across 64 database servers, they reduced query latency from 800ms to 15ms and eliminated downtime during traffic spikes.

Database Sharding Learning Path

flowchart LR
  A[Database Design] --> B[Database Indexing]
  B --> C[Database Sharding]
  C --> D[Distributed Systems]
  C:::current
  classDef current fill:#f90,color:#fff,stroke:#333,stroke-width:2px

What Is Database Sharding?

Sharding splits a large dataset into smaller chunks called shards. Each shard is an independent database instance containing a subset of the data. Together, all shards hold the complete dataset.

flowchart TD
    Client --> Router[Query Router]
    Router --> Shard1[Shard 1
Users A-H]
    Router --> Shard2[Shard 2
Users I-P]
    Router --> Shard3[Shard 3
Users Q-Z]
    Shard1 --> DB1[(Database 1)]
    Shard2 --> DB2[(Database 2)]
    Shard3 --> DB3[(Database 3)]

Sharding is distinct from Replication: Replication copies all data to multiple servers (for read scaling), while sharding splits data across servers (for write scaling and data volume).

Sharding Strategies

Range Sharding

Range sharding divides data based on a range of values in the shard key.

-- Shard 1: user_ids 1 - 1,000,000
CREATE TABLE users_shard_1 (LIKE users INCLUDING ALL);
INSERT INTO users_shard_1 SELECT * FROM users WHERE user_id BETWEEN 1 AND 1000000;

-- Shard 2: user_ids 1,000,001 - 2,000,000
CREATE TABLE users_shard_2 (LIKE users INCLUDING ALL);
INSERT INTO users_shard_2 SELECT * FROM users WHERE user_id BETWEEN 1000001 AND 2000000;

-- Shard 3: user_ids 2,000,001 - 3,000,000
CREATE TABLE users_shard_3 (LIKE users INCLUDING ALL);
INSERT INTO users_shard_3 SELECT * FROM users WHERE user_id BETWEEN 2000001 AND 3000000;

# Application-level routing based on range
def get_shard_for_user(user_id: int) -> str:
    if 1 <= user_id <= 1_000_000:
        return "shard_1"
    elif 1_000_001 <= user_id <= 2_000_000:
        return "shard_2"
    else:
        return "shard_3"

Expected behavior: Each query routes to exactly one shard based on the user_id range. Range shards are simple to implement but can create hotspots if data distribution is uneven.

Pros: Simple, range scans are efficient, easy to implement. Cons: Can cause hotspot shards, requires manual rebalancing when data grows.

Consistent Hashing

Consistent hashing maps each shard key to a position on a hash ring. Each shard covers a range of the ring. When a shard is added or removed, only a fraction of keys need to move.

import hashlib

class ConsistentHashRing:
    def __init__(self, nodes: list[str], replicas: int = 100):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            self.add_node(node)

    def add_node(self, node: str):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            self.sorted_keys.append(key)
        self.sorted_keys.sort()

    def remove_node(self, node: str):
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            del self.ring[key]
            self.sorted_keys.remove(key)

    def get_node(self, key: str) -> str:
        if not self.ring:
            return None
        hash_key = self._hash(key)
        for node_key in self.sorted_keys:
            if hash_key <= node_key:
                return self.ring[node_key]
        return self.ring[self.sorted_keys[0]]

    @staticmethod
    def _hash(key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

# Usage: shard user data across 4 nodes
ring = ConsistentHashRing(["shard_0", "shard_1", "shard_2", "shard_3"])
shard = ring.get_node("user_12345")
print(f"User 12345 goes to {shard}")

Expected output:

User 12345 goes to shard_2

Pros: Minimal data movement during scaling, even distribution. Cons: More complex, range queries across shards are inefficient.

Directory-Based Sharding

A lookup table (or distributed key-value store) maps each shard key to its shard location.

-- The shard directory table (stored separately, often in a metadata database)
CREATE TABLE shard_directory (
    shard_key VARCHAR(255) PRIMARY KEY,
    shard_id INT NOT NULL,
    shard_host VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Query to find which shard holds a specific user
SELECT shard_id, shard_host
FROM shard_directory
WHERE shard_key = 'user_12345';

# Application-level directory lookup
from redis import Redis

cache = Redis(host='localhost', port=6379)

def get_shard_for_user(user_id: str) -> str:
    # Try cache first (fast path)
    shard = cache.get(f"shard:{user_id}")
    if shard:
        return shard.decode()

    # Fall back to directory database
    shard = lookup_in_directory_db(user_id)
    cache.setex(f"shard:{user_id}", 3600, shard)  # Cache for 1 hour
    return shard

Expected behavior: Directory lookups add one hop per query but provide complete flexibility. Hot shards can be rebalanced by updating the directory.

Pros: Maximum flexibility, easy rebalancing. Cons: Single point of failure (directory itself), extra lookup latency.

Geo-Sharding

Shards are placed in different geographic regions, and data is distributed based on user location.

-- Shard assigned by geographic region
CREATE TABLE user_shard_map (
    user_id BIGINT PRIMARY KEY,
    region VARCHAR(10) NOT NULL,  -- 'US', 'EU', 'ASIA'
    shard_id INT NOT NULL
);

-- Route US users to the US shard
-- Route EU users to the EU shard
-- Route ASIA users to the ASIA shard

-- Query within a region is fast (local)
SELECT * FROM us_users WHERE user_id = 12345;

Expected behavior: Users in Europe query the EU shard (low latency), while US users query the US shard. Cross-region queries are slow but rare.

Choosing a Shard Key

The shard key determines how data is distributed. A poor shard key causes hotspots and uneven distribution.

-- Good shard key: high cardinality, even distribution
CREATE TABLE orders_sharded (
    order_id BIGINT NOT NULL,
    user_id BIGINT NOT NULL,
    -- shard key should have high cardinality
    PRIMARY KEY (order_id, user_id)  -- user_id is the shard key
);

-- Bad shard key: low cardinality causes hotspots
-- If you shard by status, the 'active' shard gets 80% of writes
-- Shard key: status (active, inactive, suspended) -- BAD IDEA

Cross-Shard Queries

Queries that need data from multiple shards are expensive. The query router must fan out to all shards and merge results.

# Cross-shard query: find top 10 users by score across all shards
import asyncio
import asyncpg

async def query_all_shards(query: str, shards: list[str]):
    """Execute a query on all shards and merge results."""
    async def query_shard(shard_dsn: str):
        conn = await asyncpg.connect(shard_dsn)
        results = await conn.fetch(query)
        await conn.close()
        return results

    # Fan out to all shards in parallel
    tasks = [query_shard(shard) for shard in shards]
    shard_results = await asyncio.gather(*tasks)

    # Merge and sort results
    all_results = []
    for results in shard_results:
        all_results.extend(results)
    return sorted(all_results, key=lambda r: r['score'], reverse=True)[:10]

Expected behavior: Cross-shard queries work but are slower. The application must handle merging, sorting, and pagination across shards.

Resharding

As data grows, existing shards become too large. Resharding redistributes data across more shards.

flowchart LR
  subgraph Before
    S1[Shard 1
0-1M users]
    S2[Shard 2
1M-2M users]
    S3[Shard 3
2M-3M users]
  end
  subgraph After
    N1[Shard 1
0-500K users]
    N2[Shard 2
500K-1M users]
    N3[Shard 3
1M-1.5M users]
    N4[Shard 4
1.5M-2M users]
    N5[Shard 5
2M-2.5M users]
    N6[Shard 6
2.5M-3M users]
  end
  S1 --> N1
  S1 --> N2
  S2 --> N3
  S2 --> N4
  S3 --> N5
  S3 --> N6

Strategies for resharding:

• Offline migration: Stop writes, move data, restart. Simple but causes downtime.
• Virtual shards: Use more shards than servers from the start. Move virtual shards between servers without data redistribution.
• Dual writes: Write to old and new shards simultaneously during migration.

Common Sharding Errors

1. Choosing a Low-Cardinality Shard Key

Sharding by status (3 values) creates 3 shards where one handles 90% of traffic. All benefits of sharding are lost.

2. Ignoring Cross-Shard Join Costs

JOINs across shards are extremely slow. Design your schema so that related data lives on the same shard (e.g., user data and user's orders share the same shard key).

3. Not Planning for Resharding

Your data will grow. If you don't plan for resharding from day one, you'll face painful downtime later. Use consistent hashing or virtual shards to make resharding manageable.

4. Relying on Auto-Increment IDs Across Shards

Each shard generates its own auto-increment IDs, causing collisions. Use UUIDs, Snowflake IDs, or shard-aware ID generators.

-- WRONG: Auto-increment across shards causes duplicate IDs
-- Shard 1 inserts: id=1, Shard 2 inserts: id=1 → CONFLICT

-- RIGHT: Use UUIDs or shard-prefixed IDs
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    -- Or use a composite key
    shard_id INT NOT NULL,
    local_id BIGINT NOT NULL,
    PRIMARY KEY (shard_id, local_id)
);

5. Making Cross-Shard Transactions

Distributed transactions across shards are slow, complex, and error-prone. Design your data model to avoid them.

6. Ignoring Hotspot Shards

Some users generate 100x more traffic than others. Monitor shard load and consider sub-sharding hot users or using directory-based sharding for flexibility.

7. Forgetting About Backup and Restore

Each shard needs its own backup strategy. Restoring a single shard should not require restoring the entire cluster.

Practice Questions

1. What is the difference between sharding and Replication?

Sharding splits data across servers (each server has a subset). Replication copies all data to every server (each server has a complete copy). Sharding scales writes; Replication scales reads.

2. Why is consistent hashing better than modulo-based sharding for scaling?

With modulo sharding (user_id % N), adding or removing a node causes almost all keys to remap. Consistent hashing moves only K/N keys when a node changes, where K is total keys and N is node count.

3. What makes a good shard key?

High cardinality (many distinct values), even data distribution, and alignment with query patterns so that most queries hit a single shard.

4. How do you handle cross-shard queries?

Route the query to all shards, collect results, and merge them in the application layer. Avoid cross-shard JOINs by co-locating related data on the same shard.

5. Challenge: Design a sharding strategy for a global messaging app.

Your messaging app has 500 million users sending 10 billion messages daily. Users message mostly within their geographic region. Answer: Use geo-sharding as the primary strategy (messages stored in the user's region). Within each region, use consistent hashing on user_id. Co-locate a user's messages, contacts, and profile on the same shard. Use a directory service for user-to-shard mapping. Cross-region messages are delivered via async Replication. Use virtual shards for easy resharding.

FAQ

Try It Yourself

Set up a sharded database with PostgreSQL and Citus:

1. Install Citus extension on three PostgreSQL instances
2. Configure one coordinator and two worker nodes
3. Create a distributed table using create_distributed_table('orders', 'user_id')
4. Insert 1 million rows and observe data distribution
5. Run queries that hit a single shard and queries that span shards
6. Add a third worker and observe rebalancing

What's Next

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.