Database Sharding in System Design: Strategies, Shard Keys and Examples

Database sharding is a technique for splitting one large database into several smaller databases, called shards, that each hold a portion of the rows and run on separate servers. Together the shards hold all the data, but no single server has to store everything or handle every query. Sharding is how applications with hundreds of millions of users keep their databases fast.

This tutorial explains what sharding is and when you need it, how it differs from vertical partitioning and replication, how to choose a shard key, the main sharding strategies, consistent hashing, how queries find the right shard, the hard problems sharding creates, and how to implement shard routing and unique IDs in Java.

1. What Is Database Sharding?

Imagine a library that has grown so large that one building can no longer hold all the books, and the queue at the single front desk stretches around the block. The solution is to open several branches: books by authors A to H go to Branch 1, I to P to Branch 2, and Q to Z to Branch 3. Each branch is smaller, faster to search and has its own desk.

Sharding does the same for a database. A users table with 500 million rows might be split across ten shards with about 50 million rows each. Each shard:

• has the same schema (the same tables and columns),
• holds a different subset of rows,
• runs on its own server with its own CPU, memory and disk,
• is usually replicated for availability.

A shard key, such as user_id, decides which shard each row belongs to. Sharding is a form of horizontal scaling: you add more machines instead of buying a bigger one.

2. Why Shard a Database?

Most applications should not shard on day one. A single well-tuned database server with read replicas can handle a surprising amount of traffic. Sharding becomes necessary when one of these limits is reached:

• Data size: the data no longer fits comfortably on one server's disk, or the working set no longer fits in memory.
• Write throughput: replicas only scale reads. Every write still goes to one primary, and eventually that primary cannot keep up.
• Vertical scaling ceiling: the biggest affordable server is not big enough, or upgrading further becomes very expensive.
• Operational pain: backups, restores, index builds and schema changes on one huge database take hours.
• Data locality: keeping European users' data in Europe for latency or legal reasons.

In return, sharding gives you more total storage, more write capacity, smaller indexes, faster maintenance per shard and fault isolation: if one shard fails, only the users on that shard are affected.

3. Vertical Partitioning vs Horizontal Sharding

There are two ways to split a table:

• Vertical partitioning splits a table by columns. Frequently used columns stay in one table and rarely used or large columns move to another. Every row still exists in both tables, linked by the same ID.
• Horizontal partitioning splits a table by rows. Each partition has all the columns but only some of the rows. When those partitions live on different servers, it is called sharding.

The term partitioning is sometimes used for splitting a table inside one database server (for example, PostgreSQL table partitioning by month), while sharding means the pieces live on different servers. The ideas are the same; the difference is where the pieces are stored.

4. Choosing a Shard Key

The shard key is the single most important sharding decision, because it is very hard to change later. A good shard key has these properties:

• High cardinality: many distinct values, so data can be spread across many shards. A country column with 30 values cannot spread data across 100 shards.
• Even distribution: values should be spread evenly, so no shard gets far more data than others.
• Even access: traffic should be spread too. A shard key that puts a celebrity's millions of followers on one shard creates a hotspot.
• Matches common queries: most queries should include the shard key, so they go to exactly one shard instead of all of them.
• Stable: the value should not change. If the shard key changes, the row must move to another shard.

5. Sharding Strategies

Once you have a shard key, a sharding strategy decides which shard each key value maps to.

Range-based sharding

Each shard holds a continuous range of key values, such as user IDs 1–1,000,000 on Shard 1 and 1,000,001–2,000,000 on Shard 2. It is easy to understand and makes range queries efficient. The risk is uneven load: if keys increase over time, every new row lands on the last shard, which becomes a hotspot while older shards sit idle.

Hash-based sharding

The shard is calculated from a hash of the key, for example shard = hash(user_id) % number_of_shards. A good hash function scatters keys evenly, so data and traffic are balanced. The downsides: range queries must ask every shard, and changing the number of shards changes the result of the modulo for most keys, forcing a huge data migration. Consistent hashing, covered next, solves that second problem.

Directory-based (lookup) sharding

A separate lookup table records which shard holds each key or group of keys, for example "tenant 812 → Shard 4". It is the most flexible approach: you can move a busy tenant to its own shard just by updating one row. The cost is an extra lookup on every query (usually cached), and the directory itself must be highly available, because if it goes down, nobody can find their data.

Geographic sharding

Data is split by location: European users on servers in Europe, Indian users on servers in India. This lowers latency and helps meet data residency laws. Shards can become unbalanced if one region grows much faster than others, so it is often combined with hash sharding inside each region.

6. Consistent Hashing for Resharding

With hash(key) % N, going from 4 shards to 5 changes the answer for about 80% of keys, so most of your data must move at once. Consistent hashing avoids this.

Picture the full range of hash values bent into a circle, a hash ring. Each shard is placed on the ring at the position of its own hash. To find a key's shard, hash the key to a point on the ring and walk clockwise until you reach a shard.

When a shard is added, it takes over only the keys between itself and the previous shard on the ring. When a shard is removed, its keys move to the next shard clockwise. On average, only about 1/N of keys move.

With only a few points on the ring, some shards would get much bigger arcs than others. The fix is virtual nodes: each physical shard is placed on the ring many times (for example 256 times) under different names, so its share evens out. Consistent hashing is used by Cassandra, DynamoDB-style stores, Memcached client libraries and many load balancers.

7. How Queries Find the Right Shard

Something must translate "user 1042" into "Shard 2". There are three common places to put this logic:

• Application-level routing: the application code (or a library inside it) computes the shard and opens a connection to it. Simple and fast, but every service that touches the data needs the same routing logic.
• Proxy or middleware: the application talks to a proxy that looks like a normal database, and the proxy routes queries. Examples include Vitess for MySQL, Citus for PostgreSQL and Apache ShardingSphere. Applications stay simpler.
• Database-native sharding: the database handles routing itself. MongoDB uses mongos routers and config servers; Cassandra and DynamoDB route by partition key automatically; distributed SQL databases such as CockroachDB split and move data ranges on their own.

A query that includes the shard key goes to exactly one shard. A query without it must be sent to every shard and the results combined, which is called scatter-gather. Scatter-gather queries get slower as you add shards, so keep them rare.

8. Challenges of Sharding

Sharding solves scale problems but introduces new ones. Understand them before you commit:

• Cross-shard queries and joins: joining data that lives on different shards cannot be done by the database. You must query several shards and join in the application, duplicate (denormalise) data, or design shard keys so related data lives together.
• Cross-shard transactions: an update touching two shards is no longer a simple local transaction. You need two-phase commit, sagas, or a design that avoids such updates.
• Hotspots: one popular key (a celebrity account, a huge customer) can overload its shard. Fixes include splitting hot keys, adding a random suffix to spread them, or moving them to a dedicated shard.
• Resharding: moving data when shards fill up is risky and must happen while the system is live. It needs careful tooling: copy data, keep it in sync, switch traffic, then clean up.
• Unique IDs: auto-increment counters work per shard, so two shards would both create order number 1001. You need globally unique IDs.
• Unique constraints across shards: the database cannot enforce "email must be unique" across shards. A separate lookup table or service is needed.
• Operational complexity: schema changes, backups, monitoring and upgrades must be repeated for every shard and kept in sync.
• Reporting: company-wide analytics need data from all shards, so teams usually copy data into a separate data warehouse.

9. Generating Unique IDs Across Shards

A popular solution, inspired by the "Snowflake" approach, packs three parts into a single 64-bit number: a timestamp, a shard (or machine) ID and a per-millisecond sequence. IDs are unique without any coordination, roughly sorted by time, and even tell you which shard created them.

package com.javabytechie.sharding;

import java.time.Instant;

/**
 * 64-bit IDs: 41 bits time | 10 bits shard id | 12 bits sequence.
 * IDs are unique across shards and roughly sorted by creation time.
 */
public class ShardAwareIdGenerator {

    private static final long CUSTOM_EPOCH = Instant.parse("2025-01-01T00:00:00Z").toEpochMilli();
    private static final int SHARD_BITS = 10;
    private static final int SEQUENCE_BITS = 12;
    private static final long MAX_SEQUENCE = (1L << SEQUENCE_BITS) - 1;   // 4095

    private final long shardId;
    private long lastMillis = -1;
    private long sequence = 0;

    public ShardAwareIdGenerator(int shardId) {
        if (shardId < 0 || shardId >= (1 << SHARD_BITS)) {
            throw new IllegalArgumentException("shardId must be between 0 and 1023");
        }
        this.shardId = shardId;
    }

    public synchronized long nextId() {
        long now = System.currentTimeMillis();
        if (now < lastMillis) {
            throw new IllegalStateException("Clock moved backwards");
        }
        if (now == lastMillis) {
            sequence = (sequence + 1) & MAX_SEQUENCE;
            if (sequence == 0) {                       // 4096 IDs used this millisecond: wait
                while (now <= lastMillis) {
                    now = System.currentTimeMillis();
                }
            }
        } else {
            sequence = 0;
        }
        lastMillis = now;
        return ((now - CUSTOM_EPOCH) << (SHARD_BITS + SEQUENCE_BITS))
                | (shardId << SEQUENCE_BITS)
                | sequence;
    }

    public static int shardOf(long id) {
        return (int) ((id >> SEQUENCE_BITS) & ((1 << SHARD_BITS) - 1));
    }

    public static void main(String[] args) {
        ShardAwareIdGenerator shard3 = new ShardAwareIdGenerator(3);
        ShardAwareIdGenerator shard7 = new ShardAwareIdGenerator(7);

        for (int i = 0; i < 2; i++) {
            long a = shard3.nextId();
            long b = shard7.nextId();
            System.out.println("id " + a + " -> created on shard " + shardOf(a));
            System.out.println("id " + b + " -> created on shard " + shardOf(b));
        }
    }
}

id 237723593104961536 -> created on shard 3
id 237723593104977920 -> created on shard 7
id 237723593125933056 -> created on shard 3
id 237723593125949440 -> created on shard 7

With 41 bits of milliseconds, this layout lasts for about 69 years from the custom epoch. Each shard can create 4,096 IDs per millisecond. Other options include UUIDs (simple but large and random, which hurts B-tree index locality; time-ordered UUID version 7 helps), or a central ID service that hands out blocks of IDs.

10. Implementing Shard Routing in Java

The following class implements a consistent hash ring with virtual nodes. Its main method compares how many keys must move when a fifth shard is added, using simple modulo versus consistent hashing:

package com.javabytechie.sharding;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Map;
import java.util.TreeMap;

public class ConsistentHashRing {

    private final TreeMap<Long, String> ring = new TreeMap<>();
    private final int virtualNodesPerShard;

    public ConsistentHashRing(int virtualNodesPerShard) {
        this.virtualNodesPerShard = virtualNodesPerShard;
    }

    public void addShard(String shard) {
        for (int i = 0; i < virtualNodesPerShard; i++) {
            ring.put(hash(shard + "#" + i), shard);
        }
    }

    public void removeShard(String shard) {
        for (int i = 0; i < virtualNodesPerShard; i++) {
            ring.remove(hash(shard + "#" + i));
        }
    }

    public String shardFor(String key) {
        // Walk clockwise to the first virtual node at or after the key's position
        Map.Entry<Long, String> entry = ring.ceilingEntry(hash(key));
        return (entry != null ? entry : ring.firstEntry()).getValue();
    }

    static long hash(String value) {
        try {
            byte[] digest = MessageDigest.getInstance("MD5").digest(value.getBytes(StandardCharsets.UTF_8));
            return ByteBuffer.wrap(digest).getLong();   // first 8 bytes as a signed long
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException(e);
        }
    }

    static String moduloShard(String key, int shardCount) {
        return "shard-" + Math.floorMod(hash(key), shardCount);
    }

    public static void main(String[] args) {
        int keys = 100_000;

        // 1. Simple modulo: hash(key) % N
        int movedModulo = 0;
        for (int i = 0; i < keys; i++) {
            String key = "user-" + i;
            if (!moduloShard(key, 4).equals(moduloShard(key, 5))) {
                movedModulo++;
            }
        }

        // 2. Consistent hashing with 256 virtual nodes per shard
        ConsistentHashRing ring = new ConsistentHashRing(256);
        for (int s = 0; s < 4; s++) {
            ring.addShard("shard-" + s);
        }
        String[] before = new String[keys];
        Map<String, Integer> load = new TreeMap<>();
        for (int i = 0; i < keys; i++) {
            before[i] = ring.shardFor("user-" + i);
            load.merge(before[i], 1, Integer::sum);
        }
        System.out.println("Keys per shard (4 shards): " + load);

        ring.addShard("shard-4");
        int movedRing = 0;
        for (int i = 0; i < keys; i++) {
            if (!ring.shardFor("user-" + i).equals(before[i])) {
                movedRing++;
            }
        }

        System.out.printf("hash %% N           : adding a 5th shard moves %,d of %,d keys (%.1f%%)%n",
                movedModulo, keys, 100.0 * movedModulo / keys);
        System.out.printf("consistent hashing : adding a 5th shard moves %,d of %,d keys (%.1f%%)%n",
                movedRing, keys, 100.0 * movedRing / keys);
    }
}

Keys per shard (4 shards): {shard-0=24301, shard-1=24012, shard-2=26137, shard-3=25550}
hash % N           : adding a 5th shard moves 80,021 of 100,000 keys (80.0%)
consistent hashing : adding a 5th shard moves 20,889 of 100,000 keys (20.9%)

The result matches the theory: modulo hashing reshuffles about four out of five keys, while consistent hashing moves only about one in five, which is the share the new shard should own. The virtual nodes keep each shard's load close to 25%.

11. Routing to Shards in Spring Boot

In a Spring Boot application with application-level sharding, you can hold the current shard in a thread-local context and let an AbstractRoutingDataSource pick the matching connection pool:

package com.javabytechie.sharding;

public final class ShardContext {

    private static final ThreadLocal<String> CURRENT = new ThreadLocal<>();

    private ShardContext() { }

    public static void set(String shard) { CURRENT.set(shard); }
    public static String get()           { return CURRENT.get(); }
    public static void clear()           { CURRENT.remove(); }
}

package com.javabytechie.sharding;

import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;

public class ShardRoutingDataSource extends AbstractRoutingDataSource {

    @Override
    protected Object determineCurrentLookupKey() {
        return ShardContext.get();   // e.g. "shard-2"; register one DataSource per shard under these keys
    }
}

package com.javabytechie.sharding;

import org.springframework.stereotype.Component;

@Component
public class OrderFacade {

    private final ConsistentHashRing ring;
    private final OrderService orderService;   // its methods are @Transactional

    public OrderFacade(ConsistentHashRing ring, OrderService orderService) {
        this.ring = ring;
        this.orderService = orderService;
    }

    public Order placeOrder(long customerId, OrderRequest request) {
        ShardContext.set(ring.shardFor("customer-" + customerId));   // choose the shard first
        try {
            return orderService.createOrder(customerId, request);    // transaction opens on that shard
        } finally {
            ShardContext.clear();                                     // never leak to the next request
        }
    }
}

Set the shard before the transactional method starts, because the connection is chosen when the transaction begins. Always clear the context in a finally block, since server threads are reused for other requests. For larger systems, a sharding proxy or a database with native sharding usually saves you from writing and maintaining this plumbing.

12. Databases with Built-in Sharding

• MongoDB: native sharding with a chosen shard key, range or hashed, automatic chunk balancing and mongos query routers.
• Cassandra and ScyllaDB: data is always partitioned by the partition key across a consistent hash ring with virtual nodes.
• Amazon DynamoDB: partitions data by partition key automatically and splits partitions as they grow.
• Vitess: a sharding layer for MySQL, originally built to scale a very large video platform.
• Citus: an extension that distributes PostgreSQL tables across worker nodes.
• CockroachDB, YugabyteDB, Google Spanner: distributed SQL databases that split data into ranges and rebalance them automatically while keeping ACID transactions.
• Elasticsearch: splits each index into a fixed number of primary shards with replicas.

13. Sharding vs Partitioning vs Replication

In practice these are combined: each shard is a replicated group (a primary with replicas), and tables inside a shard may be partitioned by date.

14. Before You Shard: Try These First

Because sharding adds permanent complexity, exhaust simpler options first:

1. Optimise queries and indexes. Many "we need to shard" problems are really missing indexes or inefficient queries.
2. Add caching with Redis or a CDN to absorb repeated reads.
3. Add read replicas for read-heavy workloads.
4. Scale up to a larger server; modern machines are very powerful.
5. Partition large tables within one server, and archive old data.
6. Split by feature (functional partitioning): move a large, independent area such as analytics or messaging into its own database.

When these are no longer enough, and especially when write volume or data size outgrows a single primary, it is time to shard.

15. Common Mistakes to Avoid

• Sharding too early, before simpler options are exhausted, and paying the complexity cost for years.
• Choosing a monotonically increasing shard key like a timestamp with range sharding, which sends every new write to one shard.
• Choosing a low-cardinality key like country or status, which cannot spread data evenly.
• Ignoring query patterns, so most queries become slow scatter-gather operations.
• Using hash % N with no resharding plan, making it painful to add shards later.
• Relying on auto-increment IDs, causing duplicate keys across shards.
• Forgetting that each shard needs replicas and backups; a shard without a replica is a single point of failure for its users.

16. Best Practices

• Pick a shard key with high cardinality, even distribution and presence in most queries.
• Keep data that is read or written together on the same shard.
• Use consistent hashing with virtual nodes, or a directory, so shards can be added with limited data movement.
• Start with more logical shards than physical servers (for example 256 logical shards on 8 servers) so you can rebalance by moving whole logical shards.
• Use globally unique IDs from the start.
• Replicate every shard and back it up independently.
• Monitor size, traffic and latency per shard to spot hotspots early.
• Automate schema migrations so every shard stays identical.
• Prefer managed or native sharding solutions over building your own when possible.

17. Summary

• Sharding splits rows across multiple database servers so data size and write traffic can grow beyond one machine.
• The shard key decides where each row lives; it should have high cardinality, spread data and traffic evenly and match common queries.
• Strategies include range, hash, consistent hashing, directory and geographic sharding, each with different trade-offs.
• Sharding creates new challenges: cross-shard joins and transactions, hotspots, resharding, unique IDs and more operational work.
• Shard only after indexing, caching, replicas and vertical scaling are no longer enough, and combine sharding with replication for availability.

Done well, sharding lets a database grow almost without limit; done hastily, it creates problems that last for years. Open the FAQ tab for answers to common questions about database sharding.

Frequently Asked Questions (FAQ)