Sharding — Splitting the Data

What is it? #

Sharding splits your data across several databases, each holding a portion of the rows.

Replication helps with reads. Sharding is what helps with writes and with data too large for one machine, because each shard handles only its own slice.

Everything depends on the shard key — the column that decides which shard a row belongs to. Choose it well and queries stay fast; choose it badly and you get hot shards and cross-shard queries.

It should be a last resort. Sharding makes joins, transactions, unique constraints and reporting much harder, and undoing it is painful.

Think of it like this #

One library growing too large for its building. You split it across several branches — A–F here, G–M there, and so on.

Each branch is smaller and faster to search. But "how many books do we have in total" now means asking every branch, and moving a book between branches is a real operation rather than shifting a shelf.

Simple example #

A messaging application writes millions of messages a day. One database cannot keep up. Sharding by conversation ID keeps every message of a conversation together, so the common query hits exactly one shard.

Code #

Sharding strategies

Range        shard = by id range (1-1M, 1M-2M)
             simple, but new data all lands on the last shard — a hot spot

Hash         shard = hash(key) % shard_count
             even spread, but changing shard_count reshuffles almost everything

Consistent   hash onto a ring
hashing      adding a shard moves only a small fraction of keys

Directory    a lookup table maps key → shard
             flexible, but the lookup becomes a dependency and a bottleneck

import hashlib


class ShardRouter:
    def __init__(self, pools: list):
        self.pools = pools                      # one connection pool per shard

    def shard_for(self, key: str):
        digest = hashlib.sha256(key.encode()).hexdigest()
        return self.pools[int(digest, 16) % len(self.pools)]

    def messages_for(self, conversation_id: str, limit: int = 50):
        pool = self.shard_for(conversation_id)  # one shard only
        with pool.connection() as conn:
            return conn.execute(
                "SELECT * FROM messages WHERE conversation_id = %s "
                "ORDER BY created_at DESC LIMIT %s",
                (conversation_id, limit),
            ).fetchall()

    def total_messages(self) -> int:
        # Cross-shard query: ask every shard and combine. Slow, avoid in hot paths.
        total = 0
        for pool in self.pools:
            with pool.connection() as conn:
                total += conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
        return total

What sharding costs you

joins            across shards: not possible in the database, do it in the app
transactions     across shards: no single ACID transaction
unique keys      globally unique IDs must be generated, not auto-incremented
reporting        every aggregate becomes a fan-out query
rebalancing      adding a shard means moving data while staying online
operations       backups, migrations and monitoring, multiplied

How it works #

The shard key decides everything. Choosing conversation_id means all messages in a conversation live together, so the most common query — recent messages in one conversation — touches exactly one shard.

Hashing the key spreads data evenly. The weakness is that % shard_count changes when you add a shard, remapping almost every key. Consistent hashing solves this by moving only a small fraction.

Range sharding is simpler to reason about but produces hot spots for sequential keys: sharding by auto-increment ID sends all new writes to the last shard, which is the opposite of what you want.

total_messages shows the real cost. Any question that is not scoped by the shard key must be asked of every shard and combined in application code. Aggregates that were one SQL query become fan-out operations.

Globally unique IDs need a different approach, since each shard's auto-increment starts at 1. UUIDs or a distributed ID scheme such as Snowflake are the usual answers.

Uneven distribution is the other practical hazard. A celebrity account or a huge tenant can make one shard far busier than the rest, which is why key choice must consider real-world distribution and not just theoretical uniformity.

Real-world use #

Very large systems shard: social networks by user ID, messaging by conversation, multi-tenant SaaS by tenant ID. In the SaaS case the shard key is obvious and queries are naturally scoped, which makes it far less painful.

Before sharding, the usual ladder is: add indexes, cache aggressively, add read replicas, move heavy columns or tables elsewhere, archive old data, and buy a bigger machine. Modern hardware is large, and most applications never need to go further.

Some databases handle it for you. Vitess, Citus and several managed cloud databases distribute data across nodes while presenting a single interface — considerably less work than sharding in application code.

The strongest practical advice is to choose the shard key deliberately and early if you know you will need it. Changing it later means migrating every row.

Common mistakes #

• Sharding before exhausting indexes, caching, replicas and vertical scaling.
• Choosing a shard key that makes common queries cross shards.
• Sharding by sequential ID, so all new writes hit one shard.
• Relying on auto-increment IDs, which collide across shards.
• Forgetting that reporting and admin queries become fan-out operations.

Practice #

Design the sharding for a multi-tenant application where each company has its own users, projects and tasks. Choose a shard key, list the three most common queries and confirm each hits one shard, then identify one query that would require a fan-out and how you would avoid it.