Introduction

Databases have natural limits. A single server fits only so much data and handles only so many writes. When you exceed these limits, you distribute the load across multiple servers.

Sharding works when your data naturally partitions. User data partitions by user ID. Order data partitions by order ID. The shard key determines which server stores each record.

graph LR
    Users[Users Table] -->|user_id % 4| Shard0[(Shard 0)]
    Users -->|user_id % 4| Shard1[(Shard 1)]
    Users -->|user_id % 4| Shard2[(Shard 2)]
    Users -->|user_id % 4| Shard3[(Shard 3)]

This approach assumes data access follows the partition key. A user record is accessed using the user ID. Orders for that user are accessed using order ID plus user ID. As long as queries include the shard key, routing stays simple.

Sharding Tools Landscape

Different tools sit at different layers of the sharding stack. Choose based on how much control you need:

Tool	Type	Sharding Approach	Best For
Vitess	Middleware (MySQL)	Vertical sharding + horizontal sharding	MySQL workloads needing scale, YouTube-scale deployments
PlanetScale	Managed Vitess	Horizontal sharding via Vitess	teams wanting MySQL compatibility without managing Vitess
CockroachDB	Distributed SQL	Automatic range-based sharding	Global applications needing strong consistency
Spanner	Distributed SQL	Automatic sharding, TrueTime	Large-scale global applications with budget for managed solution
YugabyteDB	Distributed SQL	Hash and range sharding	PostgreSQL/Cassandra compatibility with scale
Citus	PostgreSQL extension	Hash-based sharding	Teams on PostgreSQL needing distributed scale
KScale	Managed sharding	Automatic sharding	Teams wanting managed sharding without database changes

Vitess was built at YouTube to scale MySQL beyond what a single MySQL instance could handle. It handles connection pooling, query routing, and resharding automatically. The tradeoff is operational complexity. Vitess itself requires expertise to run.

CockroachDB and Spanner both present a single logical database while sharding automatically underneath. The difference is consistency model: CockroachDB uses HLC-based consistency while Spanner uses TrueTime hardware. Spanner is more expensive but handles global writes with less latency due to its TrueTime guarantees.

For most teams, starting with a distributed database like CockroachDB or YugabyteDB eliminates the need for manual sharding entirely. The database handles distribution while you work with normal SQL.

Database-Specific Sharding Configuration

Each database handles sharding differently. Understanding the native options helps you decide whether to use built-in sharding or a middleware approach.

MongoDB Sharding

MongoDB provides built-in sharding with automatic chunk distribution:

// Enable sharding on the database
sh.enableSharding("myapp");

// Shard a collection by hashed shard key (for even distribution)
sh.shardCollection("myapp.orders", { order_id: "hashed" });

// Shard by range (when range queries matter)
sh.shardCollection("myapp.events", { timestamp: 1 });

// Check cluster status
sh.status();

// Manually split a chunk to enable migration
db.adminCommand({
  split: "myapp.orders",
  middle: { order_id: NumberLong(500000) },
});

// Move a chunk to a specific shard
db.adminCommand({
  moveChunk: "myapp.orders",
  find: { order_id: NumberLong(500000) },
  to: "shard2",
});

Shard key selection for MongoDB:

Use Case	Shard Key	Why
User data	user_id (hashed)	Even distribution, user-specific queries efficient
Time-series events	device_id + timestamp (compound)	Query by device range, timestamps within device
E-commerce	user_id + order_id (compound)	User orders together, order lookups by user
Multi-tenant SaaS	tenant_id (hashed)	Tenant isolation, even load per tenant

PostgreSQL Citus

Citus extends PostgreSQL for distributed tables. It partitions tables across worker nodes while presenting a single PostgreSQL interface:

-- Install Citus extension on coordinator
CREATE EXTENSION IF NOT EXISTS citus;

-- Add worker nodes
SELECT citus_add_node('worker-1.internal', 5432);
SELECT citus_add_node('worker-2.internal', 5432);
SELECT citus_add_node('worker-3.internal', 5432);

-- Create distributed table (by hash)
CREATE TABLE users (
    user_id BIGSERIAL,
    email TEXT NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now(),
    PRIMARY KEY (user_id)
);
SELECT create_distributed_table('users', 'user_id', shard_count => 16);

-- Create reference table (replicated to all nodes)
CREATE TABLE countries (
    country_code TEXT PRIMARY KEY,
    name TEXT NOT NULL
);
SELECT create_reference_table('countries');

-- Check shard placement
SELECT * FROM citus_shards;
SELECT * FROM citus_shard_placements;

Citus supports co-located tables, where joined data stays on the same shard, and distributed tables that span shards. The create_distributed_table function takes the shard key and number of shards as arguments.

CockroachDB

CockroachDB shards automatically by default. You control data distribution through range splitting and leaseholder placement:

-- Split a table into ranges by primary key
ALTER TABLE users SPLIT AT VALUES ('m'), ('t');

-- Move a range to a specific store
ALTER TABLE users EXPERIMENTAL_RELOCATE VALUES (1, 'store-1');

-- Check range status
SELECT * FROM crdb_internal.ranges_no_leases;

-- Set up locality-aware replication
ALTER TABLE users SET locality_aware_replication = true;

CockroachDB’s KV layer automatically splits ranges when they exceed the target size (default 512 MiB). Forcing specific placement is rarely needed. The optimizer handles most distribution decisions.

When to Use Built-In vs Middleware Sharding

Approach	Examples	Best For
Built-in database sharding	MongoDB sharding, CockroachDB, Spanner	When the database handles distribution natively
Extension-based sharding	Citus for PostgreSQL, YugabyteDB	When you want PostgreSQL compatibility with scale
Middleware sharding	Vitess, PlanetScale	When you need MySQL compatibility with Vitess routing
Application-level sharding	Custom shard router	When no other option fits and you control all queries

Operational Procedures

Adding a Shard

Adding a shard with consistent hashing is the safest approach. The ring structure limits redistribution to neighboring keys:

def add_shard_with_consistent_hashing(router, new_shard_id, virtual_nodes=100):
    """
    Add a new shard to an existing consistent hash ring.
    Only keys in the new shard's range are redistributed.
    """
    # Record current state
    current_distribution = {
        shard: count_keys(router, shard)
        for shard in router.shards
    }

    # Add new shard to ring
    router.add_shard(new_shard_id, virtual_nodes)

    # Verify redistribution is limited to adjacent range
    new_distribution = {
        shard: count_keys(router, shard)
        for shard in router.shards
    }

    # Calculate migration percentage
    total_keys = sum(current_distribution.values())
    migrated = sum(abs(new_distribution.get(s, 0) - current_distribution.get(s, 0))
                   for s in set(new_distribution) | set(current_distribution)) // 2

    migration_pct = migrated / total_keys * 100
    print(f"Migration: {migration_pct:.1f}% of keys moved")

    return migration_pct < 10  # Expect <10% migration with consistent hashing

For MongoDB, adding a shard triggers automatic rebalancing of chunks across available shards. The balancer runs in the background and moves chunks gradually to avoid impacting performance.

For Citus, adding a worker node requires rebalancing distributed tables:

-- Add new worker
SELECT citus_add_node('worker-4.internal', 5432);

-- Rebalance distributed tables (runs in background)
SELECT citus_rebalance_start();

Removing a Shard

Shard removal requires migrating data before decommissioning. Never remove a shard until all data has moved:

def remove_shard(router, shard_to_remove):
    """
    Safely remove a shard by migrating data first.
    """
    # 1. Stop routing new traffic to shard
    router.deactivate_shard(shard_to_remove)

    # 2. Migrate all data
    for key in router.get_keys_for_shard(shard_to_remove):
        new_shard = router.get_shard(key)  # Will route to new owner
        data = router.get_data(shard_to_remove, key)
        new_shard.put(key, data)

    # 3. Verify migration complete
    remaining = router.count_keys(shard_to_remove)
    if remaining > 0:
        raise ValueError(f"Shard still has {remaining} keys")

    # 4. Remove from ring
    router.remove_shard(shard_to_remove)

In MongoDB, use mongosh to migrate chunks off a shard before removing it. Always verify chunk counts and document migration before decommissioning.

Shard Migration Playbook

Live shard migration involves four phases: setup, dual-write, cutover, and cleanup. Each phase has specific validation gates.

Phase 1 — Setup:

1. Provision the new shard and verify it is healthy
2. Update the routing layer to recognize the new shard
3. Enable replication from source shard to new shard
4. Verify initial sync is complete before proceeding

def migration_phase1_setup(router, source_shard, new_shard):
    """Phase 1: Initialize migration infrastructure."""
    # Verify new shard is reachable and healthy
    if not new_shard.is_healthy():
        raise MigrationError("New shard failed health check")

    # Configure replication from source to target
    router.configure_replication(source_shard, new_shard)

    # Verify initial sync catches up
    if not new_shard.is_synced(source_shard):
        raise MigrationError("Initial sync incomplete")

    print(f"Phase 1 complete: {source_shard} → {new_shard} replication active")

def migration_phase2_dual_write(router, source_shard, new_shard):
    """Phase 2: Enable dual-write to both shards."""
    router.enable_dual_write(source_shard, new_shard)

    # Verify writes go to both
    test_key = f"__migration_test_{new_shard.id}__"
    router.write(test_key, {"test": True}, dual_write=True)

    if not new_shard.has_key(test_key):
        raise MigrationError("Dual-write verification failed")

    router.delete(test_key)
    print(f"Phase 2 complete: dual-write active for {new_shard.id}")

def migration_phase3_cutover(router, old_shard, new_shard):
    """Phase 3: Switch routing to new shard, stop writes to old."""
    # Verify all data migrated via checksum
    if not router.verify_checksum(old_shard, new_shard):
        raise MigrationError("Checksum mismatch — migration incomplete")

    # Switch routing
    router.switch_shard(old_shard, new_shard)
    router.disable_write(old_shard)

    print(f"Phase 3 complete: cutover to {new_shard.id}")

def migration_phase4_cleanup(router, old_shard, new_shard):
    """Phase 4: Remove old shard, verify new shard is healthy."""
    # Verify no traffic going to old shard
    if router.count_requests(old_shard) > 0:
        raise MigrationError("Old shard still receiving traffic")

    # Remove from ring
    router.remove_shard(old_shard)

    # Verify new shard handles all traffic
    if not new_shard.is_healthy():
        raise MigrationError("New shard unhealthy after cleanup")

    print(f"Phase 4 complete: {old_shard.id} removed, {new_shard.id} active")

def run_migration_playbook(router, source_shard, new_shard):
    """Execute all four phases of shard migration."""
    migration_phase1_setup(router, source_shard, new_shard)
    migration_phase2_dual_write(router, source_shard, new_shard)
    migration_phase3_cutover(router, source_shard, new_shard)
    migration_phase4_cleanup(router, source_shard, new_shard)
    print("Migration playbook complete")

Validation gates per phase:

Phase	Gate
Phase 1 — Setup	New shard healthy, initial sync complete
Phase 2 — Dual Write	Verification write confirmed on new shard
Phase 3 — Cutover	Checksum matches, zero divergence
Phase 4 — Cleanup	Old shard has zero traffic, removed from ring

Security checklist for migration operations:

• Implement authentication at both router and shard level
• Use TLS encryption for all router-to-shard communication
• Apply least privilege for router database users
• Audit log all cross-shard operations
• Encrypt data at rest on each shard
• Implement network segmentation between shards
• Secure shard-to-shard replication channels
• Regular security reviews of routing logic
• Implement access controls for rebalancing operations
• Backup encryption keys separately from shard data

Best Practices Summary

Design Phase:

• Choose high-cardinality shard keys with even write distribution
• Model access patterns before selecting shard key — >80% of queries should filter by the key
• Test distribution with production-like data before going live
• Start with consistent hashing rather than modulo hashing

Migration Phase:

• Use virtual nodes for gradual, low-impact resharding
• Implement dual-write during migration to maintain consistency
• Validate checksum before cutover; verify no divergence between old and new shards
• Set per-shard timeouts to prevent slow shards from blocking scatter-gather queries

Operations Phase:

• Monitor per-shard latency, storage, and connection pool utilization
• Alert on shard distribution skew >2x and storage >80% capacity
• Maintain replication factor >= 2 per shard for fault tolerance
• Plan shard split before hitting limits — do not wait for crisis

Anti-Patterns to Avoid:

• Sharding before exhausting simpler scaling options (vertical, replicas, caching)
• Using timestamp or date-based keys for high-write workloads
• Cross-shard transactions as a primary pattern — redesign schema instead
• Assuming linear scalability — adding shards gives diminishing returns

Real-World Case Studies

Instagram’s Sharding Evolution

Instagram’s user database started on a single PostgreSQL server. As they grew, they added read replicas, then started partitioning. Their sharding evolution went through three generations.

First generation was pure UUID-based sharding: user_id determined the shard via modulo. Simple but inflexible — adding shards required changing the modulo divisor, which meant remapping every user to a new shard. They managed this with a migration that ran for months while serving live traffic.

Second generation moved to consistent hashing with virtual nodes. Adding a shard now moved only the neighboring keys instead of remapping all users. But they still had a problem: their shard key was user_id alone, and Instagram’s user distribution was not uniform. Some users were more active than others, and some users had relationships with much larger audiences, creating write hotspots for specific shards.

Third generation introduced application-level sharding: the application layer decided not just which shard a user’s data lived on, but also cached user data aggressively and used a fanout-on-read model for celebrity activity feeds. Write amplification for celebrity posts was managed by separating celebrity posts from regular user posts — celebrity posts were sharded differently (by post_id) than regular posts (by user_id).

The lesson: Instagram’s biggest challenge was not distributing data evenly but managing the cross-shard queries that their feed and notification systems required. They solved this with denormalization — pre-computing follower feeds and storing them close to the users who needed them. The shard key was not just a storage decision, it was an access pattern decision.

Real-World Failure Scenarios

Understanding how sharding systems fail in production helps you design more resilient architectures. These scenarios are drawn from real incidents and common pitfalls.

Hot Shard Due to Hash Collision

A ride-sharing company used driver_id as the shard key with hash-based partitioning. All drivers were assigned sequential IDs during onboarding. When their batch import process ran, it created thousands of drivers with consecutive IDs. Since hash(driver_id) mapped sequential IDs to the same hash ring position, writes from the batch job all landed on a single shard, creating a write hot spot that caused 40x latency spikes during import windows.

Fix: Add entropy to the shard key — use hash(driver_id + random_salt) or a different hash function that spreads sequential IDs across the ring. Never assume IDs have uniform distribution.

Router Table Corruption During Reshard

A social media company running on Vitess performed a resharding operation. During the migration, a network partition interrupted the routing table update. Some Vitess tablets received the new routing configuration while others did not. Requests for migrated keys were routed to old shards that no longer held the data, resulting in missing posts and conversations for affected users for 3 hours before the issue was identified.

Fix: Use a two-phase routing update: (1) deploy new routing rules that check both old and new shard locations, (2) verify all keys migrated via checksum, (3) switch to new routing once migration is 100% complete. Implement circuit breakers that fail closed (return error) for keys that should be on a new shard but are not found.

Cross-Shard JOIN Timeout Cascade

A gaming company used a microservices architecture with a sharded PostgreSQL cluster. One service executed a report that required a JOIN across all shards. The query coordinator sent sub-queries to 8 shards. One shard had a slow disk due to a failed drive that was being rebuilt — it took 45 seconds to respond while others responded in under 100ms. The query coordinator had no per-shard timeout, so the entire report request timed out after 60 seconds, consuming connection slots on all shards and causing cascading timeouts for other services.

Fix: Implement per-shard query timeouts. If a shard exceeds its timeout, fail the request with a partial result error rather than blocking indefinitely. Monitor per-shard latency in the routing layer and mark slow shards as unhealthy until the issue is resolved.

Schema Migration on Sharded Database

A fintech company needed to add a new column to a table spanning 16 shards. They used ALTER TABLE directly on each shard during low-traffic windows. However, one shard had a significantly larger table due to uneven data distribution. The ALTER took 45 minutes on that shard while others completed in under 5 minutes. During the migration window, the application code that expected both schemas (old and new) ran in production — but the large shard was still running the old schema while smaller shards had the new schema. Queries that joined data across shards failed because of type mismatches on the new column.

Fix: For sharded schema migrations, run ALTERs in stages: (1) add the column as nullable with no NOT NULL constraint, (2) deploy application code that handles both schemas, (3) backfill default values for existing rows on each shard, (4) add NOT NULL constraint once all rows have valid values. Use pt-online-schema-change or similar tools that handle backfill automatically. Always test schema changes on the largest shard first.

Shard Rebalancing Data Loss

A distributed key-value store using consistent hashing performed a resharding operation. The rebalance process migrated keys from old shards to new shards. During migration, a write arrived for a key that had been partially migrated — the key existed on both old and new shards with different values. The router, configured to prefer the new shard, returned stale data to the user. When the user updated their data, the update went to the new shard while the old shard still held the previous version. When the old shard was decommissioned, the newer version was lost.

Fix: Use a write-lock during migration for active keys: either block writes to keys being migrated, or use a versioning scheme that lets you detect and resolve conflicts. Implement dual-write during migration so both shards see all writes, and only decommission the old shard once all writes for the migration window have been verified on the new shard.

Trade-off Analysis

Sharding involves fundamental trade-offs across multiple dimensions. Understanding these helps you make informed architectural decisions.

Scenario	Modulo Sharding	Consistent Hashing	Distributed SQL (CockroachDB/Spanner)
Shard addition	100% data migration	~33% migration (doubling scenario)	Automatic, ~0% application migration
Operational complexity	Low (simple math)	Medium (ring management)	Low (database handles it)
Hot spot risk	High (no virtual nodes)	Low (virtual nodes even distribution)	Medium (automatic but unaware of access patterns)
Cross-shard query performance	Same as consistent hashing	Scatter-gather required	Optimizer handles, but distributed JOINs are expensive
Consistency model	Depends on database	Typically eventual	Strong (Spanner TrueTime) or eventual (CockroachDB follower reads)
Cost	Low (uses standard DB)	Medium (custom routing layer)	High (managed service or complex self-hosted)
Schema changes	Complex across shards	Complex across shards	Online schema changes supported
Write latency	Lowest (single shard)	Low (single shard per key)	Higher (distributed consensus required)
Read latency	Lowest	Low (local replica option)	Variable (follower reads vs primary)

When to Choose Each Approach

Modulo sharding is acceptable for write-once read-many workloads where data is never re-sharded. It is the wrong choice for any production system expecting growth. If you are using modulo, plan your migration path to consistent hashing before you need it.

Consistent hashing is the right choice when you need fine-grained control over shard placement, you expect to add and remove shards frequently, or you want to minimize migration percentage when resharding. The tradeoff is building or maintaining a routing layer.

Distributed SQL is the right choice when you want to focus on application logic rather than database operations, you need strong consistency across global deployments, and your team lacks distributed systems expertise. The tradeoff is higher latency for distributed transactions and higher cost for managed solutions.

Shard Key Cardinality Trade-offs

Shard Key Type	Max Shards Possible	Hot Spot Risk	Range Query Support
tenant_id (10 tenants)	10	High if tenant sizes differ	Efficient within tenant
user_id (millions of users)	Limited by hash space	Low	Requires scatter-gather
order_id (GUID/UUID)	Virtually unlimited	Very low	Requires scatter-gather
tenant_id + entity_id (composite)	Limited by tenant cardinality	Medium	Efficient within tenant + entity
hashed(user_id)	Virtually unlimited	Very low	Random distribution, no range support

Interview Questions

Further Reading

Papers:

• “Dynamo: Amazon’s Highly Available Key-value Store” — foundational paper on consistent hashing and eventual consistency in production systems
• “Bigtable: A Distributed Storage System for Structured Data” — sharding model behind Google’s database infrastructure
• “Spanner: Google’s Globally Distributed Database” — TrueTime and global consistency guarantees

Books:

• Designing Data-Intensive Applications by Martin Kleppmann — comprehensive coverage of partitioning, replication, and consistency models
• Database Internals by Alex Petrov — deep dive into storage engines, B-trees, and LSM trees underlying distributed databases

Tools and References:

• Vitess Documentation — production-grade MySQL sharding middleware
• CockroachDB Architecture — distributed SQL internals
• MongoDB Sharding Documentation — MongoDB’s chunk-based sharding
• Citus Sharding Guide — PostgreSQL extension for distributed tables

Related Topics:

• Database Scaling — scaling strategies beyond sharding
• Consistent Hashing — hash ring algorithms and virtual nodes
• Distributed Systems Fundamentals — consistency models, consensus, and fault tolerance

Conclusion

| Use this checklist when designing or reviewing a sharding strategy.

Shard Key Selection: high cardinality, even write distribution, queries include the key, no monotonic patterns.

Consistent Hashing: use virtual nodes, plan for resharding before hitting limits, verify migration stays under 10% when adding shards.

Cross-Shard Query Design: denormalize to avoid JOINs across shards, per-shard timeouts, circuit breakers per shard.

Operations: monitor per-shard latency and storage, alert on skew over 2x and storage over 80%, maintain replication factor at least 2, test shard operations in staging first.