Home/Learn/System Design/Database Sharding

Database Sharding

Intermediate

Data Management

Sharding partitions data across multiple database instances based on a shard key. It enables horizontal scaling of the data tier but introduces complexity in queries, joins, and rebalancing.

Overview

Database sharding (horizontal partitioning) splits a large dataset across multiple independent database instances, each holding a subset of the data. Each shard is a fully functional database. A shard key (e.g. user_id, tenant_id) determines which shard stores a given row. Sharding enables horizontal scaling beyond the limits of a single machine — each shard handles a fraction of the total reads and writes. However, sharding introduces significant complexity: cross-shard queries require scatter-gather, joins across shards are expensive or impossible, rebalancing data when adding shards is operationally complex, and maintaining ACID across shards requires distributed transactions. Common strategies include range-based sharding (e.g. user_id 1-1M → shard 1), hash-based sharding (hash(user_id) % N), and directory-based sharding (a lookup service maps keys to shards).

Sharding Strategies

Range-based sharding is simple but can create hotspots. Hash-based sharding distributes evenly but makes range queries hard. Directory-based sharding is flexible but adds a lookup overhead.

Conceptual + Java — sharding strategies

// 1. Range-based sharding
// user_id 1–1,000,000      → Shard A
// user_id 1,000,001–2,000,000 → Shard B
// Pros: range queries within a shard are fast
// Cons: hotspots if users are not uniformly distributed

// 2. Hash-based sharding
// shard = hash(user_id) % num_shards
// Pros: even distribution
// Cons: range queries require scatter-gather across all shards
//       adding a shard requires rehashing (use consistent hashing!)

// 3. Directory-based sharding
// Lookup table: { user_id: shard_id }
// Pros: maximum flexibility, easy rebalancing
// Cons: lookup service is a single point of failure and bottleneck

// Shard routing in application code
public class ShardRouter {
    private final int numShards;

    public DataSource getShardFor(String userId) {
        int shard = Math.abs(userId.hashCode()) % numShards;
        return shardDataSources.get(shard);
    }
}

Cross-Shard Queries

Queries that span multiple shards require scatter-gather — the application sends the query to all relevant shards and aggregates results. This is slower and more complex than single-shard queries.

Conceptual — cross-shard queries and denormalisation

// Scatter-gather pattern for cross-shard query
//
// App: "find all orders with total > $100"
//   ├── Query Shard 1: SELECT * FROM orders WHERE total > 100
//   ├── Query Shard 2: SELECT * FROM orders WHERE total > 100
//   └── Query Shard 3: SELECT * FROM orders WHERE total > 100
//   └── Merge + sort results in application layer
//
// Performance: O(num_shards) latency (parallelised)
// Complexity: pagination, sorting, aggregation across shards

// Best practice: design shard key to keep related data together
// Good: shard by tenant_id → all tenant data on one shard
// Bad:  shard by order_id → user's orders spread across shards

// Denormalisation: duplicate frequently joined data
// Instead of cross-shard join: orders ⟗ users
// Store user_name directly in orders table (denormalised)

Resharding & Rebalancing

Adding or removing shards requires moving data between shards. Consistent hashing minimises data movement. Production resharding uses double-write or change-data-capture migration strategies.

Conceptual — resharding strategies

// Resharding with consistent hashing
//
// Without consistent hashing:
//   shard = hash(key) % 3   →  adding 4th shard moves ~75% of data!
//
// With consistent hashing:
//   Only ~1/N of keys move when adding Nth node
//   Virtual nodes ensure even distribution

// Migration strategy: dual-write approach
// 1. Start writing to BOTH old shard and new shard
// 2. Backfill: copy existing data from old → new shard
// 3. Verify consistency
// 4. Switch reads to new shard
// 5. Stop writing to old shard
// 6. Clean up old shard

// Vitess (YouTube's MySQL sharding layer)
// Handles shard routing, resharding, and schema changes
// Used by: YouTube, Slack, Square, GitHub

Key Points to Remember

1Sharding splits data across multiple DB instances based on a shard key — enabling horizontal data scaling.
2Hash-based sharding distributes evenly; range-based enables efficient range queries; directory-based is most flexible.
3Choose shard keys that keep related data together to minimise cross-shard queries.
4Cross-shard joins and transactions are expensive — prefer denormalisation and eventual consistency.
5Use consistent hashing to minimise data movement when adding shards.
6Tools like Vitess, Citus, and ProxySQL simplify sharding for MySQL and PostgreSQL.

Interview Questions

What is database sharding and why is it needed?

EasyTCS

Compare range-based vs hash-based sharding.

MediumAmazon

How do you handle cross-shard queries efficiently?

MediumGoogle

How would you add a new shard with zero downtime?

HardUber

Design a sharding strategy for a multi-tenant SaaS platform.

HardAtlassian

Ask Aria about Database Sharding

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Latency vs Throughput

Database Replication