Database Sharding 101: When and How to Scale Horizontally

· database sharding horizontal scaling distributed systems scalability database architecture

Database Sharding 101: When and How to Scale Horizontally

Imagine your application is growing fast. You’ve optimized queries, added caching layers, and even upgraded to the most expensive database instance your cloud provider offers. Yet, your database is still struggling. Queries are slow, write operations are queuing up, and you’re starting to see timeout errors in production. You’ve hit the vertical scaling wall.

This is where database sharding comes in. Sharding is a horizontal scaling strategy that distributes data across multiple database instances, each called a shard. It’s not a silver bullet, but when applied correctly, it can unlock near-linear scalability. In this post, I’ll walk through when sharding makes sense, how to implement it, and what pitfalls to watch out for.

What Is Database Sharding?

At its core, sharding is a database partitioning technique where you split a large dataset into smaller, independent chunks and store each chunk on a separate database server. Each shard holds a subset of the data, and together they form the complete dataset.

Unlike replication, where every node has a copy of the same data, sharding ensures that each node has a unique slice of the data. This reduces the load on any single server and allows the system to scale horizontally by adding more shards.

Sharding vs. Partitioning

It’s important to distinguish sharding from other forms of partitioning:

Sharding is a specific form of horizontal partitioning that implies distribution across physical or virtual machines.

When Should You Consider Sharding?

Sharding adds complexity to your architecture. You should only consider it when you’ve exhausted simpler alternatives. Here are the signs that it might be time:

  1. Data size exceeds a single server’s capacity – Your dataset is growing beyond what one machine can store, even with compression and archiving.
  2. Write throughput is bottlenecked – Your application has high write volume that a single primary database cannot handle, even with read replicas.
  3. Query latency is unacceptable – Even with indexing and query optimization, response times are too high due to the sheer volume of data.
  4. Geographic distribution requirements – You need data to be close to users in different regions to reduce latency.

Before sharding, try these alternatives:

If you’ve tried all these and still hit limits, sharding might be your next step.

Sharding Strategies

Choosing how to distribute data across shards is the most critical design decision. Here are the common strategies:

1. Range-Based Sharding

Data is partitioned based on a range of values in a shard key, such as user ID ranges or date ranges.

Example: Shard 1 stores users with IDs 1–100,000. Shard 2 stores IDs 100,001–200,000, and so on.

Pros:

Cons:

2. Hash-Based Sharding

A hash function is applied to the shard key (e.g., hash(user_id) % N where N is the number of shards). This distributes data more uniformly.

Example: Using user_id mod 4 to distribute data across 4 shards.

1
2
3
public int getShardId(Long userId, int totalShards) {
    return (int) (userId % totalShards);
}

Pros:

Cons:

3. Directory-Based Sharding

A lookup service (shard map) maintains a mapping between shard keys and shards. This decouples the shard assignment from the data.

Example: A configuration table in a metadata database maps customer_id to shard 3.

Pros:

Cons:

4. Geographic Sharding

Data is partitioned based on geographic region. This is common for applications with users spread across the world.

Example: European users’ data on servers in Frankfurt, US users on servers in Virginia.

Pros:

Cons:

Choosing a Shard Key

The shard key is the column or set of columns used to determine which shard a row belongs to. A good shard key should:

Common shard keys include:

Implementing Sharding: A Practical Example

Let’s say we have a social media application with a posts table that’s growing too large. We decide to shard by user_id using hash-based sharding.

Step 1: Set Up Multiple Database Instances

We create three MySQL databases: shard_0, shard_1, and shard_2.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# docker-compose.yml for local development
version: '3.8'
services:
  shard-0:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: social_app
    ports:
      - "3306:3306"

  shard-1:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: social_app
    ports:
      - "3307:3306"

  shard-2:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: social_app
    ports:
      - "3308:3306"

Step 2: Create the Same Schema on Each Shard

1
2
3
4
5
6
7
8
CREATE TABLE posts (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    user_id BIGINT NOT NULL,
    title VARCHAR(255),
    content TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_id (user_id)
);

Step 3: Implement a Shard Router in the Application

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;

public class ShardRouter {

    private final Map<Integer, DataSource> shardDataSources;
    private final int totalShards;

    public ShardRouter(Map<Integer, DataSource> shardDataSources) {
        this.shardDataSources = shardDataSources;
        this.totalShards = shardDataSources.size();
    }

    public DataSource getShardForUser(long userId) {
        int shardId = (int) (userId % totalShards);
        return shardDataSources.get(shardId);
    }

    public DataSource getShardById(int shardId) {
        return shardDataSources.get(shardId);
    }
}

Step 4: Route Queries to the Correct Shard

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class PostRepository {

    private final ShardRouter shardRouter;
    private final JdbcTemplate jdbcTemplate;

    public PostRepository(ShardRouter shardRouter) {
        this.shardRouter = shardRouter;
        // Assume JdbcTemplate is created per shard
    }

    public Post findPostById(Long postId, Long userId) {
        DataSource shard = shardRouter.getShardForUser(userId);
        JdbcTemplate shardTemplate = new JdbcTemplate(shard);
        String sql = "SELECT * FROM posts WHERE id = ? AND user_id = ?";
        return shardTemplate.queryForObject(sql, new Object[]{postId, userId}, new PostRowMapper());
    }

    public List<Post> findPostsByUser(Long userId) {
        DataSource shard = shardRouter.getShardForUser(userId);
        JdbcTemplate shardTemplate = new JdbcTemplate(shard);
        String sql = "SELECT * FROM posts WHERE user_id = ? ORDER BY created_at DESC";
        return shardTemplate.query(sql, new Object[]{userId}, new PostRowMapper());
    }
}

Step 5: Handle Cross-Shard Queries

For queries that need data from multiple shards (e.g., a global feed), you must query all shards and aggregate results in the application layer.

1
2
3
4
5
6
7
8
9
10
11
12
public List<Post> getRecentPostsGlobally(int limit) {
    List<Post> allPosts = new ArrayList<>();
    for (int i = 0; i < totalShards; i++) {
        DataSource shard = shardRouter.getShardById(i);
        JdbcTemplate shardTemplate = new JdbcTemplate(shard);
        String sql = "SELECT * FROM posts ORDER BY created_at DESC LIMIT ?";
        allPosts.addAll(shardTemplate.query(sql, new Object[]{limit}, new PostRowMapper()));
    }
    // Sort and limit in application
    allPosts.sort((a, b) -> b.getCreatedAt().compareTo(a.getCreatedAt()));
    return allPosts.subList(0, Math.min(limit, allPosts.size()));
}

Common Challenges and Solutions

Challenge 1: Resharding

When you need to add or remove shards, hash-based sharding requires recalculating all mappings. This is often done by:

Challenge 2: Distributed Transactions

Transactions that span multiple shards are complex and slow. Solutions include:

Challenge 3: Join Operations

Joins across shards are expensive. Mitigations:

Challenge 4: Backup and Restore

Each shard is independent, so you need to back up each one separately. Ensure your backup strategy covers all shards and that you can restore to a consistent point in time across shards.

When NOT to Shard

Sharding is not for every situation. Avoid it if:

Alternatives to Sharding

Before committing to sharding, consider these modern alternatives:

Key Takeaways