Database Sharding at Scale: 2 Billion Rows Across 64 Shards with Online Rebalancing

When your single PostgreSQL instance hits 500M rows and query performance degrades despite every index optimization you can throw at it, you face a choice: vertical scaling (bigger machines with diminishing returns) or horizontal scaling (sharding). We chose sharding when our transaction table reached 800M rows and p99 query latency crossed 200ms — a threshold our SLA could not tolerate.

Eighteen months later, we operate 64 shards holding 2.1 billion rows, processing 45K transactions per second with p99 read latency under 12ms. This article covers the shard key decisions, the online rebalancing system that saved us from 3 AM maintenance windows, and the cross-shard query patterns that make the system usable for analytics.

When to Shard (And When Not To)

Sharding introduces significant complexity. Before committing, we exhausted simpler scaling options:

Optimization	Impact	Effort	Result
Index optimization	-40% query time	Low	Temporary — regressed in 3 months
Read replicas	-60% read load on primary	Low	Helped reads, writes still bottlenecked
Connection pooling (PgBouncer)	+3x connection capacity	Low	Delayed the problem by 2 months
Table partitioning	-50% query time on partitioned tables	Medium	Good for time-series, not our access pattern
Vertical scaling (64→128 vCPU)	+40% throughput	Low	Expensive, diminishing returns
Sharding	+64x write capacity	High	Solved the fundamental constraint

The decision criteria: if your workload is write-heavy, your data exceeds what a single machine can index efficiently, and your access patterns have a natural partition key — sharding is the correct architecture. Before reaching that point, explore optimizations like Kubernetes-level cost optimization to ensure you are getting the most from your existing infrastructure.

Shard Key Selection

The shard key determines everything: query routing, data distribution, and future rebalancing complexity. We evaluated three candidates for our financial transaction system:

Candidate Key	Distribution	Query Pattern Match	Hotspot Risk	Rebalancing Ease
transaction_id (UUID)	Perfect uniform	Poor (most queries by account)	None	Easy
account_id	Moderate skew (power law)	Excellent (90% of queries)	High (whale accounts)	Moderate
account_id + consistent hashing	Controlled distribution	Excellent	Low (hash spreads whales)	Moderate

We chose account_id with consistent hashing because 90% of our queries filter by account. This means most queries hit a single shard — no scatter-gather required.

Consistent Hashing Implementation

# Shard router using consistent hashing with virtual nodes
import hashlib
from bisect import bisect_right
from typing import List, Tuple, Optional

class ShardRouter:
    """
    Routes account_ids to shards using consistent hashing.
    Virtual nodes ensure even distribution despite power-law account sizes.
    """
    
    def __init__(self, num_shards: int = 64, virtual_nodes_per_shard: int = 256):
        self.num_shards = num_shards
        self.virtual_nodes = virtual_nodes_per_shard
        self.ring: List[Tuple[int, int]] = []  # (hash_position, shard_id)
        self._build_ring()
    
    def _build_ring(self) -> None:
        """Build the hash ring with virtual nodes for even distribution."""
        for shard_id in range(self.num_shards):
            for vnode in range(self.virtual_nodes):
                key = f"shard-{shard_id}-vnode-{vnode}"
                hash_val = self._hash(key)
                self.ring.append((hash_val, shard_id))
        
        self.ring.sort(key=lambda x: x[0])
        self._positions = [pos for pos, _ in self.ring]
        self._shards = [shard for _, shard in self.ring]
    
    def get_shard(self, account_id: str) -> int:
        """Determine which shard owns this account."""
        hash_val = self._hash(account_id)
        idx = bisect_right(self._positions, hash_val) % len(self.ring)
        return self._shards[idx]
    
    def get_shard_connection(self, account_id: str) -> str:
        """Return the connection string for the owning shard."""
        shard_id = self.get_shard(account_id)
        return f"postgresql://shard-{shard_id:02d}.db.internal:5432/transactions"
    
    def _hash(self, key: str) -> int:
        """Consistent hash function — must never change once in production."""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_rebalance_plan(self, new_num_shards: int) -> dict:
        """
        Calculate which accounts need to move when adding shards.
        Returns mapping of (from_shard, to_shard) -> estimated_rows.
        """
        new_router = ShardRouter(new_num_shards, self.virtual_nodes)
        movements = {}
        
        # In practice, we scan the shard catalog for account distributions
        # This is a simplified planning algorithm
        for old_shard in range(self.num_shards):
            for vnode in range(self.virtual_nodes):
                key = f"shard-{old_shard}-vnode-{vnode}"
                new_shard = new_router.get_shard(key)
                if new_shard != old_shard:
                    pair = (old_shard, new_shard)
                    movements[pair] = movements.get(pair, 0) + 1
        
        return movements


# Usage in query routing layer
router = ShardRouter(num_shards=64)

async def get_account_transactions(account_id: str, limit: int = 50):
    shard_id = router.get_shard(account_id)
    conn = await get_connection(shard_id)
    
    return await conn.fetch("""
        SELECT transaction_id, amount, currency, status, created_at
        FROM transactions
        WHERE account_id = $1
        ORDER BY created_at DESC
        LIMIT $2
    """, account_id, limit)

Online Rebalancing

The hardest operational challenge with sharding is rebalancing: moving data between shards without downtime. We built a custom rebalancing pipeline that performs online migration with zero downtime and zero data loss.

Rebalancing Process

Phase 1: PREPARE     → Create target schema, set up CDC pipeline
Phase 2: BACKFILL    → Copy existing data to new shard (background)
Phase 3: DUAL-WRITE  → Write to both old and new shard
Phase 4: VERIFY      → Compare checksums, validate completeness
Phase 5: CUTOVER     → Switch reads to new shard
Phase 6: CLEANUP     → Remove data from old shard (after cooling period)

Phase	Duration (64→80 shards)	Data Moved	Customer Impact
Prepare	2 minutes	0	None
Backfill	4.2 hours	380GB	None (background I/O)
Dual-write	30 minutes	N/A	+2ms write latency
Verify	45 minutes	N/A (read-only checksums)	None
Cutover	12 seconds	0	None (atomic switch)
Cleanup	2 hours	380GB deleted	None (background)

Dual-Write Implementation

# Dual-write handler during rebalance migration
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class MigrationPhase(Enum):
    NORMAL = "normal"
    DUAL_WRITE = "dual_write"
    CUTOVER = "cutover"

@dataclass
class MigrationState:
    account_id: str
    source_shard: int
    target_shard: int
    phase: MigrationPhase
    backfill_complete: bool
    verified: bool

class ShardMigrationWriter:
    """
    Handles writes during shard rebalancing.
    Ensures zero data loss during migration with dual-write pattern.
    """
    
    def __init__(self, router: ShardRouter, migration_registry: dict):
        self.router = router
        self.migrations = migration_registry  # account_id -> MigrationState
    
    async def write_transaction(self, account_id: str, transaction: dict) -> str:
        migration = self.migrations.get(account_id)
        
        if not migration or migration.phase == MigrationPhase.NORMAL:
            # No migration in progress — write to current shard
            shard = self.router.get_shard(account_id)
            return await self._write_to_shard(shard, transaction)
        
        elif migration.phase == MigrationPhase.DUAL_WRITE:
            # Write to both shards — source is authoritative
            source_result = await self._write_to_shard(
                migration.source_shard, transaction
            )
            
            # Best-effort write to target — will be reconciled if it fails
            try:
                await self._write_to_shard(
                    migration.target_shard, transaction
                )
            except Exception as e:
                await self._log_replication_gap(
                    migration, transaction, e
                )
            
            return source_result
        
        elif migration.phase == MigrationPhase.CUTOVER:
            # Target shard is now authoritative
            return await self._write_to_shard(
                migration.target_shard, transaction
            )
    
    async def _write_to_shard(self, shard_id: int, transaction: dict) -> str:
        conn = await get_connection(shard_id)
        result = await conn.fetchval("""
            INSERT INTO transactions (account_id, amount, currency, status, metadata)
            VALUES ($1, $2, $3, $4, $5)
            RETURNING transaction_id
        """, transaction['account_id'], transaction['amount'],
            transaction['currency'], 'pending', transaction.get('metadata', {}))
        return result

Cross-Shard Queries

Most queries (90%) hit a single shard via the account_id. But analytics queries, admin dashboards, and reporting require cross-shard aggregation.

Pattern: Scatter-Gather with Streaming

# Cross-shard query execution with parallel scatter-gather
import asyncio
from typing import AsyncIterator, List

class CrossShardQueryExecutor:
    """
    Executes queries across all shards in parallel with streaming results.
    Used for analytics, admin queries, and reporting.
    """
    
    def __init__(self, num_shards: int = 64, max_parallel: int = 16):
        self.num_shards = num_shards
        self.semaphore = asyncio.Semaphore(max_parallel)
    
    async def aggregate_query(
        self, 
        query: str, 
        params: tuple = (),
        merge_fn: callable = None
    ) -> dict:
        """
        Execute aggregation query across all shards and merge results.
        """
        tasks = []
        for shard_id in range(self.num_shards):
            task = self._query_shard(shard_id, query, params)
            tasks.append(task)
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Filter failures and merge
        successful = [r for r in results if not isinstance(r, Exception)]
        failed = [r for r in results if isinstance(r, Exception)]
        
        if failed:
            # Log but don't fail — partial results are acceptable for analytics
            logger.warning(f"Cross-shard query: {len(failed)}/{self.num_shards} shards failed")
        
        if merge_fn:
            return merge_fn(successful)
        return {"results": successful, "shards_queried": len(successful)}
    
    async def _query_shard(self, shard_id: int, query: str, params: tuple):
        async with self.semaphore:
            conn = await get_connection(shard_id)
            return await conn.fetch(query, *params)

Cross-Shard Query Performance

Query Type	Shards Hit	Latency (p50)	Latency (p99)	Use Case
Account lookup	1	3ms	12ms	All user-facing queries
Global aggregation	64	180ms	450ms	Admin dashboard
Date-range report	64	2.4s	8.2s	Daily reporting
Full scan (rare)	64	12s	34s	Ad-hoc analysis

Monitoring Shard Health

Metric	Warning Threshold	Critical Threshold	Action
Row count skew (max/avg)	> 1.5x	> 2.0x	Trigger rebalance planning
Query latency p99 per shard	> 20ms	> 50ms	Investigate hotspot
Connection pool utilization	> 70%	> 90%	Scale shard resources
Replication lag (read replicas)	> 500ms	> 2s	Page on-call

Setting up proper database replication lag monitoring per shard is essential — a lagging shard replica is harder to detect than a single-database lag event. | Storage utilization per shard | > 70% | > 85% | Plan storage expansion |

Lessons Learned

Shard key selection is irreversible (practically). Changing the shard key after launch means migrating every row. We spent three weeks evaluating shard keys before committing. That investment paid for itself many times over.

Build rebalancing from day one. Even with perfect initial distribution, access patterns shift. Accounts grow, new features change query patterns, and hotspots emerge. Online rebalancing is not a nice-to-have — it is a survival tool.

Cross-shard transactions are the enemy. We redesigned our data model to eliminate cross-shard transactions entirely. If a business operation spans accounts (transfer between accounts), we use the saga pattern with eventual consistency rather than distributed transactions.

64 shards is not magic. We chose 64 because it gives us room to grow (each shard handles ~33M rows comfortably) while keeping the routing table small. The number should be a power of 2 for clean consistent hashing, and large enough to absorb 3-5 years of growth without resharding.

Conclusion

Sharding 2 billion rows across 64 shards transformed our database from a scaling ceiling into a scaling lever. The architecture handles our current 45K TPS with headroom to 200K+ TPS by adding shards. The key investments — consistent hashing for clean routing, online rebalancing for zero-downtime operations, and scatter-gather for cross-shard analytics — make the system operationally sustainable.

If you are approaching the sharding decision, remember: sharding is a one-way door. Invest heavily in shard key selection, build the operational tooling before you need it, and design your data model to minimize cross-shard dependencies. The complexity is real, but at sufficient scale, it is the only architecture that delivers both performance and linear scalability. For teams not yet at sharding scale, RDS Proxy can extend your single-database ceiling significantly through better connection management.