As your application grows, a single database instance becomes a bottleneck. Database sharding horizontally partitions data across multiple database instances, enabling massive scale. This guide covers strategies, trade-offs, and implementation patterns.
Sharding Is a Last Resort
Database sharding distributes data across multiple database instances using horizontal partitioning. It is the most complex scaling strategy and should be your last resort. Before sharding, exhaust all other options: read replicas, connection pooling optimization, query optimization, caching (Redis/CDN), and vertical scaling (bigger instance).
The sharding key is the single most important decision — it determines query performance, data distribution, and rebalancing difficulty. A poor sharding key creates hot spots where one shard handles 80% of traffic while others sit idle. Common sharding keys include user ID (most natural for SaaS), tenant ID (multi-tenant), and geographic region (latency optimization).
Resharding is the hardest problem: when you need to add shards, existing data must be rebalanced — a massive migration. Two rebalancing approaches are downtime migration (offline) and consistent hashing (minimizes moved data). Alternatives to manual sharding include Citus (PostgreSQL sharding), Vitess (MySQL sharding), CockroachDB (auto-sharding), and Spanner (auto-sharding). Sharding works at scale for companies like Instagram and Uber, but it adds enormous operational complexity.
Understanding Database Sharding
The Scaling Problem
Every database hits a ceiling eventually, and the diagram below shows why that ceiling is structural rather than incidental. On a single instance, throughput is bounded by three resources that compete for the same machine: connections, CPU, and disk I/O. Connection pools typically cap out in the tens of thousands, and each active connection consumes memory and scheduler time. CPU is consumed by query parsing, plan generation, and index maintenance, all of which grow with the working set. Disk I/O is the hardest wall because it scales with data size and index depth, not with hardware speed. A single SSD-backed instance realistically tops out around 10,000-50,000 concurrent connections and roughly 100 TB of storage before latency or cost becomes unbearable.
Vertical scaling delays but does not solve the problem. The most expensive instances on the market cost six figures and still represent a single point of failure—a hardware fault takes the whole service down regardless of how much you paid. Replication solves the availability problem but not the scale problem: read replicas spread read traffic, yet writes and the storage ceiling remain pinned to one primary. The diagram captures this bottleneck explicitly, because recognizing that you have hit the wall is the first step toward choosing the right way over it.
┌─────────────────────────────────────────────────────────────────┐
│ Single Database Scaling Limits │
│ │
│ Requests │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ App │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ Database│ ◄── Single point of failure │
│ └────┬────┘ - Vertical scaling has limits │
│ │ - Connection pool exhaustion │
│ ▼ - I/O bandwidth saturation │
│ ┌─────────┐ │
│ │ SSD │ │
│ └─────────┘ │
│ │
│ Typical limits: │
│ - 10,000-50,000 connections max │
│ - ~100TB on single instance │
│ - Vertical scaling: $100K+ for top specs │
└─────────────────────────────────────────────────────────────────┘
The takeaway from this diagram is that sharding is fundamentally a hardware strategy: you are trading a single large machine for many smaller ones and accepting the coordination cost that comes with it. Everything that follows in this article—shard keys, routing, cross-shard queries, rebalancing—exists because this diagram is true.
Sharding Solution
The second diagram shows the shape of the answer. A shard router sits between the application and the database fleet, computing which shard owns a given row from a shard key. The router can live in the application process, as it does in the application-level sharding examples later, or it can be an external service like Vitess’ VTGate or a Citus coordinator that intercepts queries transparently. The key property to notice is that the router makes a pure function of the key: given the same key, it always returns the same shard, and that determinism is what keeps reads and writes consistent.
Four benefits fall out of this design, and it is worth being precise about which ones you actually need. Linear scaling means you can add shards to absorb more throughput, but only if your workload is key-scoped—global operations still hit every shard. Reduced per-instance load is real, because each shard only handles a fraction of the connections and data. Geographic distribution lets you place shards near their users, at the cost of making cross-region queries expensive. Isolation allows you to keep a noisy tenant or a huge log table from degrading everyone else, which is often the least-appreciated benefit of the architecture.
┌─────────────────────────────────────────────────────────────────┐
│ Horizontal Sharding Architecture │
│ │
│ Requests │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ App │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────┐ │
│ │ Shard Router │ │
│ │ (or Application Logic) │ │
│ └─────────────────────────────────────────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Shard 1 │ │ Shard 2 │ │ Shard 3 │ │ Shard N │ │
│ │ (Users │ │ (Users │ │ (Orders)│ │ (Logs) │ │
│ │ A-M) │ │ N-Z) │ │ │ │ │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│ │
│ Benefits: │
│ ✓ Linear scaling with more shards │
│ ✓ Reduced load per database │
│ ✓ Geographic distribution possible │
│ ✓ Isolation for different data types │
└─────────────────────────────────────────────────────────────────┘
Shard Key Selection
The shard key is the one decision that touches every other part of a sharded system, so it deserves its own section before we look at implementations. A good shard key has two properties: it distributes data evenly across shards, and it aligns with the way your application actually queries the data. The code in this section is analysis code—it helps you choose a key with evidence rather than intuition, then shows the classic good and bad candidates so you can recognize them quickly in your own schema.
Types of Shard Keys
The dictionary below catalogs the four families of shard keys that cover nearly every production system. Range-based keys partition the key space into contiguous intervals, which makes range scans fast but risks hot spots when writes concentrate at one end of the range, as they do with auto-incrementing IDs. Hash-based keys apply a hash function and modulo the number of shards, giving statistically even distribution at the price of scattering any range query across all shards. Directory-based keys defer the mapping to a lookup service, buying flexibility—you can rebalance without touching data—but introducing a new single point of failure in the lookup service itself. Geographic keys optimize for latency by co-locating data with its users, but they make any cross-region operation slow and complicated.
Each family is a trade-off between three axes: distribution quality, query locality, and operational flexibility. Hash keys win on distribution, range keys win on locality, directory keys win on flexibility, and geographic keys win on latency for a specific class of workload. There is no universal best family, which is exactly why this decision deserves a structured analysis rather than a guess.
SHARD_KEY_TYPES = {
"range_based": {
"description": "Partition by ranges of the key",
"example": "users_00: IDs 1-1M, users_01: IDs 1M-2M",
"pros": ["Simple to implement", "Easy range queries"],
"cons": ["Hot spots if access is not uniform"]
},
"hash_based": {
"description": "Hash function determines shard",
"example": "shard = hash(user_id) % num_shards",
"pros": ["Even distribution", "Reduced hot spots"],
"cons": ["Range queries scatter across shards"]
},
"directory_based": {
"description": "Lookup service maps keys to shards",
"example": "shard = lookup_service.get_shard(key)",
"pros": ["Flexible", "Can rebalance dynamically"],
"cons": ["Lookup service is single point"]
},
"geographic": {
"description": "Partition by geographic region",
"example": "us-east, eu-west, ap-south",
"pros": ["Low latency for regional users"],
"cons": ["Cross-region queries expensive"]
}
}
Shard Key Analysis
Reading a catalog of key types is not the same as knowing which key your system needs, so the functions below convert a candidate analysis into a quantitative decision. The outer function, analyze_shard_key_candidates, walks every column of every table in your schema and scores it against your query log. For each candidate column it computes a distribution score, which measures how evenly the hash-based shard assignments spread across shards, and a query coverage figure, which measures how often your queries filter on that column and can therefore be routed to a single shard.
The inner function, analyze_distribution, shows how the distribution score is actually derived. It replays your query log, extracts the key values from WHERE clauses, hashes them to shards, and then compares the busiest shard to the quietest. A distribution score of 1.0 means perfect evenness, and a hotspot risk above roughly 3 starts to be a practical problem because the hot shard will saturate long before the others. The important insight is that this analysis uses your real query log, not a synthetic workload, which is what makes the recommendation trustworthy.
def analyze_shard_key_candidates(schema: dict, query_log: list) -> dict:
"""Analyze potential shard keys for even distribution."""
candidates = {}
for table in schema.values():
for column in table.columns:
distribution = analyze_distribution(
query_log,
column,
table.name
)
query_patterns = analyze_query_patterns(
query_log,
column,
table.name
)
candidates[f"{table.name}.{column.name}"] = {
"distribution_score": distribution["score"],
"query_coverage": query_patterns["coverage"],
"hotspot_risk": distribution["hotspot_risk"],
"recommendation": calculate_recommendation(
distribution, query_patterns
)
}
return candidates
def analyze_distribution(query_log, column, table) -> dict:
access_counts = {}
for query in query_log:
if query.table == table and column in query.where_cols:
key_value = query.where_values[column]
shard = hash(key_value) % NUM_SHARDS
access_counts[shard] = access_counts.get(shard, 0) + 1
total = sum(access_counts.values())
distribution = [count / total for count in access_counts.values()]
return {
"score": 1 - (max(distribution) - min(distribution)),
"hotspot_risk": max(distribution) / min(distribution) if min(distribution) > 0 else float('inf'),
"distribution": distribution
}
Choosing the Right Shard Key
The last piece of the selection toolkit is pattern recognition: most good and bad shard keys fall into recognizable shapes. The GOOD_SHARD_KEYS dictionary captures the three archetypes that keep showing up in well-designed systems. user_id works whenever the vast majority of queries are scoped to a single user, which is true for social, e-commerce, and SaaS workloads—each user’s data lives on one shard, and hash distribution spreads users evenly. order_date (or any time dimension) fits time-series data, where range queries over time are the norm and range-based partitioning keeps recent data hot on the newest shard. tenant_id is the natural key for multi-tenant SaaS, giving each tenant its own home and making tenant isolation a side effect of the data layout.
The BAD_SHARD_KEYS dictionary is the more educational half, because these failure modes are subtle until you have seen them. An auto-increment ID concentrates all new writes on one shard because the newest IDs map to the same range. A status field like ‘active’ is massively skewed, so 90% of the data and traffic pile onto a single shard. A foreign key alone is useless for routing because the application does not know which shard a child row lives on without a lookup, forcing full scatter-gather for every join. In each case the fix is the same: pick a key with high cardinality, even distribution, and routing power.
GOOD_SHARD_KEYS = {
"user_id": {
"when": "Most queries are user-scoped",
"example": "SELECT * FROM orders WHERE user_id = ?",
"distribution": "hash(user_id) % N"
},
"order_date": {
"when": "Time-range queries are common",
"example": "SELECT * FROM logs WHERE date > '2026-01-01'",
"distribution": "Range-based partitioning"
},
"tenant_id": {
"when": "Multi-tenant SaaS application",
"example": "SELECT * FROM documents WHERE tenant_id = ?",
"distribution": "tenant_id % N"
}
}
BAD_SHARD_KEYS = {
"auto_increment_id": """
❌ Problem: Creates hot shard
- Newest data gets all writes
- Single shard becomes bottleneck
✅ Solution: Use composite key with random component
""",
"status_field": """
❌ Problem: Highly skewed distribution
- Most records have status='active'
- 90% of data on one shard
✅ Solution: Use user_id or date instead
""",
"foreign_key_only": """
❌ Problem: Can't route queries without the key
- Need to join across all shards
- Full scatter-gather operations
✅ Solution: Denormalize or use alternate key
"""
}
Implementation Strategies
There are two main places to put the sharding logic: inside your application or inside a database-level proxy. Both are legitimate, and the choice is more about your team’s control points than raw capability. Application-level sharding keeps the routing code in your language and in your deploy pipeline, which gives you maximum flexibility and visibility but forces every new service to reimplement routing correctly. Database-level sharding moves routing into a proxy or engine, so all services inherit it uniformly, but it constrains you to the proxy’s sharding model and adds an operational component you must run and monitor. The examples below show both, starting with the application-level approach because it makes the mechanics easiest to see.
Application-Level Sharding
The ShardRouter class below is the heart of application-level sharding: a thin, deterministic function that maps a shard key to a shard. Its constructor takes a configuration dict that declares which shard function to use, and get_shard dispatches between a hash implementation and a range implementation. Hash-based routing computes the index directly from the key, which is O(1) and perfectly even, while range routing iterates over a sorted list of ranges. The trade-off between the two is the one from the key catalog: hash spreads writes evenly but destroys range locality, range keeps locality but creates hot ranges.
The ShardedConnection class sits on top of the router and manages the per-shard connection lifecycle. get_connection lazily creates a connection the first time a shard is touched, and execute_on_shard composes routing and execution into a single call. This lazy-creation pattern matters in practice: it means the application only opens connections to shards it actually uses, and it gives you a natural seam to plug in per-shard connection pooling, retries, and observability later. The price of this simplicity is that every query must thread a shard key through the call path, and any query that does not have a key is stuck doing scatter-gather.
class ShardRouter:
def __init__(self, shard_config: dict):
self.shards = shard_config["shards"]
self.shard_function = shard_config["shard_function"]
def get_shard(self, shard_key) -> str:
if self.shard_function == "hash":
shard_index = hash(shard_key) % len(self.shards)
elif self.shard_function == "range":
shard_index = self._get_range_shard(shard_key)
else:
raise ValueError(f"Unknown function: {self.shard_function}")
return self.shards[shard_index]
def _get_range_shard(self, key) -> int:
for i, range_def in enumerate(self.shard_config["ranges"]):
if range_def["min"] <= key < range_def["max"]:
return i
return len(self.shards) - 1
class ShardedConnection:
def __init__(self, router: ShardRouter):
self.router = router
self.connections: dict[str, Connection] = {}
def get_connection(self, shard_key) -> Connection:
shard = self.router.get_shard(shard_key)
if shard not in self.connections:
self.connections[shard] = self._create_connection(shard)
return self.connections[shard]
def execute_on_shard(self, shard_key, query, params):
conn = self.get_connection(shard_key)
return conn.execute(query, params)
Sharding at the Database Level
The alternative to routing in your application is to push routing into the database layer, and the Vitess-style configuration below shows what that looks like. Vitess takes a MySQL cluster and adds a VTGate proxy that parses SQL, resolves shards through a VSchema, and routes each query to the right tablet. The SHARD_CONFIG shows the two core pieces of that model: the shard ranges, which carve the keyspace into four contiguous key ranges, and the vindex, which is Vitess’ term for a shard-key function that maps a column value to a shard.
The route_query function illustrates the crucial difference from application-level sharding: the caller does not compute anything. The application submits an ordinary parameterized SQL query, and VTGate inspects the WHERE clause, finds the user_id bind variable, runs it through the hash vindex, and delivers the query to the owning shard. This transparency is the main selling point of database-level sharding—applications keep writing normal SQL—but it is also the source of its constraints, because the proxy can only route queries that it can prove are key-scoped. Queries without a vindex column silently become scatter-gather operations, and some SQL features are unsupported or behave differently across shards.
# Vitess-style horizontal sharding (MySQL)
SHARD_CONFIG = {
"shard_ranges": [
"-4000000000000000000", # Shard 0: < -4T
"-4000000000000000000-0", # Shard 1: -4T to 0
"0-4000000000000000000", # Shard 2: 0 to 4T
"4000000000000000000-" # Shard 3: > 4T
],
"vindex": {
"user_vindex": {
"type": "hash",
"column": "user_id",
"shard_count": 4
}
}
}
# Routing query with VIndex
def route_query(vtgate_conn, query):
# VSchema tells Vitess which vindex to use
bound_query = {
"sql": "SELECT * FROM orders WHERE user_id = :user_id",
"bind_vars": {"user_id": 12345}
}
# Vitess automatically routes to correct shard
return vtgate_conn.execute(bound_query)
Range-Based Sharding
Range-based sharding is the simplest partitioner to reason about, and the RangeBasedPartitioner class shows the whole algorithm: sort the ranges, scan to find the first range that contains the key, and default to the last range if nothing matches. The sorted lookup is O(log n) if you upgrade the linear scan to a binary search, which is worth doing once the range count grows. The real appeal of this strategy shows in the LOG_PARTITIONER example: quarterly ranges for logs make each shard a self-contained time window, so pruning old data is just dropping a shard, and queries for “everything this quarter” hit exactly one shard.
Those strengths come with two well-known costs. Write hotspots appear at the boundary between ranges, because current data always falls into the newest range and that shard absorbs all of today’s writes. And hot ranges are not a transient problem—the newest shard stays hot until the range fills, then the load jumps to the next one, so you never really escape the imbalance, you just move it forward in time. Range sharding is therefore a great fit for immutable, time-bucketed data like logs and events, and a poor fit for mutable, write-hot entity data.
class RangeBasedPartitioner:
"""Partition data by key ranges."""
def __init__(self, ranges: list[tuple]):
self.ranges = sorted(ranges, key=lambda x: x[0])
def get_partition(self, key) -> int:
for i, (min_val, max_val) in enumerate(self.ranges):
if min_val <= key < max_val:
return i
return len(self.ranges) - 1
# Example: Date-based partitioning for logs
LOG_PARTITIONER = RangeBasedPartitioner([
("2025-01-01", "2025-04-01"),
("2025-04-01", "2025-07-01"),
("2025-07-01", "2025-10-01"),
("2025-10-01", "2026-01-01"),
])
def get_log_shard(timestamp) -> str:
partition = LOG_PARTITIONER.get_partition(timestamp)
return f"logs_shard_{partition}"
Consistent Hashing
Plain hash(key) % num_shards has a fatal flaw when the shard count changes: adding or
removing a single shard remaps nearly every key, triggering a full data migration.
Consistent hashing fixes this by placing both nodes and keys on a shared circular hash ring
and assigning each key to the first node it meets traveling clockwise. The
ConsistentHashRing class implements exactly that, and its most interesting design detail is
the virtual_nodes parameter set to 150. Each physical node is hashed 150 times with
different suffixes and placed on the ring 150 times, which smooths out the uneven
distribution that a small number of random ring positions would otherwise produce.
The cost model is where consistent hashing earns its keep. When one node joins or leaves the ring, only the keys whose ring position falls between the old and new node locations remap, so on average only K/N of the keys move for a K-node cluster. Compare that with a plain modulo scheme, where every key moves on every topology change. The price is a more complex implementation and the need to tune the virtual-node count to balance distribution quality against memory and lookup cost, but for any sharded system that expects to grow, the reduced rebalancing traffic is usually worth it.
import hashlib
class ConsistentHashRing:
def __init__(self, nodes: list[str], virtual_nodes: int = 150):
self.ring = {}
self.sorted_keys = []
self.virtual_nodes = virtual_nodes
for node in nodes:
self._add_node(node)
def _add_node(self, node: str):
for i in range(self.virtual_nodes):
key = self._hash(f"{node}:{i}")
self.ring[key] = node
self.sorted_keys.append(key)
self.sorted_keys.sort()
def _hash(self, key: str) -> int:
return int(hashlib.md5(key.encode()).hexdigest(), 16)
def get_node(self, key: str) -> str:
hash_key = self._hash(key)
for node_hash in self.sorted_keys:
if node_hash >= hash_key:
return self.ring[node_hash]
return self.ring[self.sorted_keys[0]]
def add_node(self, node: str):
self._add_node(node)
def remove_node(self, node: str):
for i in range(self.virtual_nodes):
key = self._hash(f"{node}:{i}")
del self.ring[key]
self.sorted_keys.remove(key)
Cross-Shard Queries
Sharding makes single-shard queries fast and simple; the hard part is what happens when a query needs data from more than one shard. Cross-shard queries come in three escalating forms: scatter-gather for aggregations, distributed joins for related data, and distributed transactions for atomic updates. Each is more expensive and more complex than the last, and an important principle of sharded design is to minimize their frequency by choosing shard keys that keep related data co-located. The three subsections below implement each form so you can see exactly what the cost is.
Scatter-Gather Pattern
The ScatterGatherQuery class implements the cheapest and most common cross-shard operation: fire the same query at every shard and merge the results. The query_all_shards method fans out with asyncio.gather, so all shards are queried concurrently and the latency of the whole operation is bounded by the slowest shard rather than the sum of all shards. Each shard runs in its own task, and per-shard failures are captured as error entries and filtered out of the result set, which means one slow or down shard degrades the answer rather than killing the request.
The aggregate_results method is where the design decisions concentrate. It concatenates the per-shard row lists and hands them to an aggregator callback, leaving the merge policy entirely to the caller. That is a deliberate choice: sorting, grouping, summing, or ranking can each be expressed as an aggregator, and keeping the combinator separate from the fetch makes the pattern reusable. What you must accept is the inherent cost—every scatter-gather touches every shard, so its cost scales with the number of shards, and it is the single most common performance trap in sharded systems.
class ScatterGatherQuery:
def __init__(self, shard_connections: dict):
self.connections = shard_connections
async def query_all_shards(self, query: str, params: dict) -> list:
"""Execute query on all shards and combine results."""
async def fetch_from_shard(shard_name, conn):
try:
result = await conn.execute(query, params)
return {"shard": shard_name, "data": result}
except Exception as e:
return {"shard": shard_name, "error": str(e)}
tasks = [
fetch_from_shard(shard, conn)
for shard, conn in self.connections.items()
]
results = await asyncio.gather(*tasks)
return [r for r in results if "error" not in r]
def aggregate_results(self, shard_results: list, aggregator: callable):
"""Combine results from all shards."""
all_data = []
for result in shard_results:
all_data.extend(result.get("data", []))
return aggregator(all_data)
Handling Joins Across Shards
Joins are where sharding hurts most, because the whole point of a join is to bring related rows together, and sharding deliberately scatters them. The ShardedJoinExecutor implements the standard four-step workaround. It fetches the left table from all relevant shards, extracts the set of join keys, fetches only the matching right-table rows, and performs the join in application memory. The in-memory join builds a hash index on the join key and walks the left rows once, which is the classic hash-join algorithm moved out of the database and into your code.
This pattern works, but its costs should shape your schema design. It ships both tables over the network, so the data volume is the product of both table sizes, and the join happens in application memory, which sets a hard ceiling on how much data can be joined at all. The pragmatic responses are denormalization—store the joined fields you need alongside the primary row so you never have to join—and key alignment, choosing shard keys so that the tables you join frequently land on the same shard. In practice, teams that shard successfully treat cross-shard joins as an exception to be designed away, not a feature to be optimized.
class ShardedJoinExecutor:
def __init__(self, router: ShardRouter):
self.router = router
async def join_across_shards(
self,
left_query,
right_query,
join_key: str,
join_type: str = "inner"
):
# Step 1: Fetch data from first table
left_shards = await self._fetch_left_table(left_query)
# Step 2: Group by join key
join_keys = set()
for row in left_shards:
join_keys.add(row[join_key])
# Step 3: Fetch matching rows from second table
right_shards = await self._fetch_right_table(
right_query,
join_keys
)
# Step 4: Perform in-memory join
right_index = {
row[join_key]: row
for row in right_shards
}
results = []
for left_row in left_shards:
key = left_row[join_key]
if key in right_index or join_type == "left":
results.append({
**left_row,
**right_index.get(key, {})
})
return results
async def _fetch_left_table(self, query):
# Execute on all relevant shards
pass
async def _fetch_right_table(self, query, keys):
# Fetch only needed keys from all shards
pass
Distributed Transactions
The hardest cross-shard problem is a transaction that must update multiple shards atomically, and the TwoPhaseCommit class shows the classic protocol. Phase 1, the prepare phase, sends each operation to its shard and asks it to stage the write and vote yes or no. Phase 2 runs only if every shard voted yes: it sends the commit and each shard finalizes. The code makes the safety property explicit—if any shard fails to prepare, the whole transaction rolls back—which is the entire point of the protocol.
What the code cannot show is why distributed transactions are so painful in practice. The protocol needs a coordinator with its own transaction log, and the coordinator becomes a single point of failure and a latency bottleneck because every operation now requires an extra round trip and the coordinator waits for the slowest shard. Worse, a crash at the wrong moment leaves shards in doubt, unable to decide whether to commit or roll back. This is why sharded architectures overwhelmingly favor avoiding distributed transactions: prefer single-shard transactions where possible, and prefer idempotent, compensatable operations (the saga pattern) over two-phase commit when atomicity across shards is truly required.
import asyncio
class TwoPhaseCommit:
def __init__(self, shard_connections: dict):
self.connections = shard_connections
async def execute_transaction(
self,
operations: list[dict]
) -> bool:
# Phase 1: Prepare
prepared = await self._prepare_phase(operations)
if not all(prepared.values()):
# Rollback on any failure
await self._rollback_phase(operations)
return False
# Phase 2: Commit
await self._commit_phase(operations)
return True
async def _prepare_phase(self, operations: list[dict]) -> dict:
results = {}
async def prepare_op(op):
shard = self.router.get_shard(op["shard_key"])
conn = self.connections[shard]
try:
await conn.execute("PREPARE TRANSACTION", op["transaction_id"])
return True
except Exception:
return False
results = await asyncio.gather(*[
prepare_op(op) for op in operations
])
return {op["transaction_id"]: r
for op, r in zip(operations, results)}
Rebalancing Shards
Rebalancing is the operation nobody wants to think about at design time and everybody has to do eventually: the moment you add shards, existing data must be redistributed so the new capacity is actually used. The sections below cover the two techniques that make rebalancing safe in production: online migration of data in batches with checkpoints, and dual-write switching so no writes are lost while data moves. Both exist because a naive “dump everything and reload” migration requires downtime that grows with your dataset, and at the scale where sharding is justified, that downtime is unacceptable.
Online Re-sharding
The ShardRebalancer class demonstrates the batch-and-checkpoint approach to moving data between a source shard and a target shard without taking either offline. The move_data loop reads a fixed-size batch of rows ordered by ID, inserts them into the target, records the last ID it saw, and repeats until the source is exhausted. Two details make this safe for production. The batch size bounds how much memory and lock time each iteration consumes, and the checkpoint_interval persists progress periodically, so if the process dies mid-migration you resume from the last checkpoint rather than restarting from zero.
The verify_integrity method highlights the hidden cost of online migration: you now have two copies of the same data, and you must prove they match. Counts are only the first check; production rebalancers also compare checksums per batch and keep the write path pointed at the source until verification passes. The design tension is everywhere: batching large improves throughput but widens the window of inconsistency, and checkpoints reduce restart cost but add write traffic. Choosing those parameters is an operational judgment informed by your data rate and your tolerance for partial divergence.
class ShardRebalancer:
def __init__(
self,
source_shard: Connection,
target_shard: Connection
):
self.source = source_shard
self.target = target_shard
async def move_data(
self,
batch_size: int = 1000,
checkpoint_interval: int = 10000
):
last_id = 0
total_moved = 0
while True:
# Read batch from source
batch = await self.source.execute("""
SELECT * FROM users
WHERE id > %s
ORDER BY id
LIMIT %s
""", (last_id, batch_size))
if not batch:
break
# Write to target
for row in batch:
await self.target.insert("users", row)
last_id = batch[-1]["id"]
total_moved += len(batch)
# Checkpoint progress
if total_moved % checkpoint_interval == 0:
await self._save_checkpoint(last_id, total_moved)
return total_moved
async def verify_integrity(self):
source_count = await self.source.count("users")
target_count = await self.target.count("users")
return source_count == target_count
Dual-Write Pattern During Migration
Dual-write is the switching mechanism that lets you cut over from an old layout to a new one with zero data loss. The DualWriteRouter tracks a migration_state that starts as “old”, moves to “migrating”, and ends as “new”. While in the migrating state, every write goes to both the old router’s shard and the new router’s shard, and every read is compared across the two—if they disagree, the router logs the mismatch and returns the new value, surfacing divergence instead of silently serving stale data. The read comparison is the key safety feature, because it turns a blind migration into an observable one.
The pattern’s strength is also its weakness: double the write traffic, and a subtle bug can cause writes to diverge permanently without being caught. That is why the migration state machine must be monotonic and externally controlled—you flip to “new” only after a verification window shows zero mismatches, and you should keep the old shards around for rollback until you are confident. In mature systems, tools like Vitess and CockroachDB implement exactly this dance internally, and understanding dual-write is what lets you operate those tools with confidence or build the same machinery by hand.
class DualWriteRouter:
def __init__(self, old_router: ShardRouter, new_router: ShardRouter):
self.old_router = old_router
self.new_router = new_router
self.migration_state = "old" # old, migrating, new
async def write(self, key, data):
# Write to both during migration
old_shard = self.old_router.get_shard(key)
new_shard = self.new_router.get_shard(key)
if self.migration_state in ["old", "migrating"]:
await self.write_to_shard(old_shard, data)
if self.migration_state in ["migrating", "new"]:
await self.write_to_shard(new_shard, data)
async def read(self, key):
# Read from both and compare during migration
old_shard = self.old_router.get_shard(key)
new_shard = self.new_router.get_shard(key)
if self.migration_state == "migrating":
old_data = await self.read_from_shard(old_shard, key)
new_data = await self.read_from_shard(new_shard, key)
if old_data != new_data:
logger.error(f"Data mismatch: {old_data} vs {new_data}")
return new_data
elif self.migration_state == "new":
return await self.read_from_shard(new_shard, key)
else:
return await self.read_from_shard(old_shard, key)
Managing Reference Data
Sharding scatters your user data across shards, but most applications also depend on a small set of reference data—country codes, currencies, timezones, subscription plans—that every shard needs to read and that almost nobody writes. This section covers the two standard ways to handle it: replicating reference tables to every shard, and serving them from a cache. The right choice depends on how often the data changes and how consistent it must be.
Distributed Reference Tables
The ReferenceDataManager class shows the cache-first strategy, and its design tells you what to cache and how long to trust it. The get_country_codes method checks Redis first, falls back to a read from any single shard, and writes the result back with a one-hour TTL. Notice the small assumptions embedded in the code: because the table is replicated, “any shard” is a safe source, and because it changes rarely, an hour of staleness is acceptable. The TTL is the explicit control knob that trades freshness against load—lower it for rapidly changing reference data, raise it for static data.
The replicate_reference_tables function is the alternative: rather than caching on demand, it truncates and reloads the reference table on every shard in one pass. This is the correct approach when the data is small, changes rarely, and every shard must see the same version immediately. The trade-off between the two strategies is classic distributed-systems tension: replication gives you immediate consistency everywhere at the cost of a full reload on every change, while caching gives you instant reads but bounded staleness. Many systems use both—replicate the truly static tables, cache the slowly-changing ones.
class ReferenceDataManager:
def __init__(self, cache: Redis):
self.cache = cache
async def get_country_codes(self) -> dict:
# Try cache first
cached = await self.cache.get("country_codes")
if cached:
return json.loads(cached)
# Fetch from any shard (replicated)
data = await self.any_shard.execute(
"SELECT code, name FROM countries"
)
result = {row["code"]: row["name"] for row in data}
# Cache with TTL
await self.cache.setex(
"country_codes",
3600,
json.dumps(result)
)
return result
# Replicate reference tables to all shards
REFERENCE_TABLES = [
"countries",
"currencies",
"timezones",
"subscription_plans"
]
def replicate_reference_tables():
for table in REFERENCE_TABLES:
source_data = master_db.fetch_table(table)
for shard in all_shards:
shard.execute(f"TRUNCATE TABLE {table}")
shard.bulk_insert(table, source_data)
Connection Pooling Per Shard
A sharded fleet multiplies the connection management problem: not one pool to size, but one pool per shard, each with its own limits and its own failure modes. The ShardedConnectionPool class models this correctly by building a separate ConnectionPool for every shard, each with its own minimum, maximum, and idle-timeout settings, rather than sharing a single global pool. That separation matters because a hot shard should not be able to exhaust connections that other shards need, and a single misbehaving shard should degrade in isolation.
The get_connection method also shows the failure behavior explicitly: when a shard’s pool is at capacity, it raises ConnectionPoolExhaustedError instead of silently queueing forever. That exception is a design statement—it forces the caller to decide what to do under load rather than letting latency degrade invisibly. In production you would typically catch it and either wait with a bounded retry or fail fast with a clear alert, and the pool would be configured so that the sum of all per-shard limits fits within your application’s own connection ceiling.
class ShardedConnectionPool:
def __init__(self, config: dict):
self.pools: dict[str, ConnectionPool] = {}
self.config = config
self._initialize_pools()
def _initialize_pools(self):
for shard_name, shard_config in self.config["shards"].items():
self.pools[shard_name] = ConnectionPool(
host=shard_config["host"],
port=shard_config["port"],
min_connections=5,
max_connections=50,
max_idle_time=300
)
async def get_connection(self, shard_key: str) -> Connection:
shard = self.router.get_shard(shard_key)
# Check connection limit per shard
if self.pools[shard].active_connections >= \
self.pools[shard].max_connections:
# Wait or throw exception
raise ConnectionPoolExhaustedError(shard)
return await self.pools[shard].acquire()
async def release_connection(self, shard_key: str, conn: Connection):
shard = self.router.get_shard(shard_key)
await self.pools[shard].release(conn)
Monitoring Sharded Systems
A sharded system has more moving parts and more ways to fail silently than any single database, which makes monitoring a first-class design concern rather than an afterthought. The metrics and health checks below are organized around a key distinction: per-shard metrics tell you whether each individual shard is healthy, while cross-shard metrics tell you whether the sharding itself is working. Both views are necessary, because a fleet of individually healthy shards can still be delivering bad results if the distribution or routing is broken.
Key Metrics
The SHARD_METRICS structure documents which numbers matter and which thresholds trigger alarms. The per-shard list covers the classic database signals—queries per second, query duration percentiles, active connections, and resource usage—because each shard is, at bottom, a database that can saturate on its own. The cross-shard list is what makes this monitoring specifically about sharding: scatter-gather query counts, distributed transaction counts and failures, rebalance progress, and replication lag. A sudden rise in scatter-gather queries usually means a new query pattern is not key-scoped, which is the most common silent performance regression in sharded systems.
The alert thresholds translate these metrics into operational rules, and two are worth explaining because they encode judgment, not just data. The shard imbalance threshold of 0.3 (30% skew) fires before a hot shard becomes a customer-facing problem, giving you time to investigate the shard key or rebalance. The connection-pool-usage threshold of 0.9 gives headroom for traffic spikes while ensuring you notice a pool approaching saturation. Thresholds like these should be treated as living configuration, tuned as your traffic profile and shard layout evolve.
SHARD_METRICS = {
"per_shard_metrics": [
"queries_per_second",
"rows_returned",
"rows_modified",
"avg_query_duration",
"p99_query_duration",
"active_connections",
"cpu_usage",
"disk_usage",
"memory_usage"
],
"cross_shard_metrics": [
"scatter_gather_queries",
"distributed_transaction_count",
"failed_transactions",
"rebalance_progress",
"replication_lag"
],
"alert_thresholds": {
"shard_imbalance": 0.3, # 30% skew triggers alert
"high_latency_p99": "500ms",
"connection_pool_usage": 0.9, # 90% triggers alert
"disk_usage": 0.85
}
}
Health Checks
Metrics give you a time series; health checks give you a binary verdict, and the
ShardHealthChecker implements the second. The check_shard_health method runs a trivial
SELECT 1 against the shard, measures its latency, pulls the pool’s active, idle, and
waiting counts, and returns a structured status report. On failure it returns the exception
message, so the difference between “shard is fine” and “shard is broken” is explicit and
machine-readable. The check_all_shards method then fans those checks out across the fleet
concurrently, which is important because a serial health check over many shards would itself
become a bottleneck during an incident.
The design choice that matters here is the use of a trivial query with full pooling
semantics. A SELECT 1 validates the connection, the pool, and the basic query path, but it
does not validate that the shard can serve real traffic or that its data is consistent—for
that you need the per-shard metrics and the cross-shard skew checks from the previous
section. Health checks and metrics are complementary: health checks drive alerting and
routing decisions, while metrics drive capacity planning and root-cause analysis. Build
both, wire both into your observability stack, and alert on the health checks first.
class ShardHealthChecker:
def __init__(self, pool_manager: ShardedConnectionPool):
self.pools = pool_manager
async def check_shard_health(self, shard: str) -> dict:
pool = self.pools.pools[shard]
try:
# Test query
start = time.time()
async with pool.acquire() as conn:
result = await conn.execute("SELECT 1")
latency = time.time() - start
# Get pool stats
stats = pool.get_stats()
return {
"shard": shard,
"healthy": True,
"latency_ms": latency * 1000,
"active_connections": stats["active"],
"idle_connections": stats["idle"],
"wait_queue": stats["waiting"]
}
except Exception as e:
return {
"shard": shard,
"healthy": False,
"error": str(e)
}
async def check_all_shards(self) -> list[dict]:
tasks = [
self.check_shard_health(shard)
for shard in self.pools.pools.keys()
]
return await asyncio.gather(*tasks)
Related Articles
Summary
Database sharding enables horizontal scaling beyond single database limits:
- Shard Key Selection is critical - choose keys that distribute load evenly and support your most common query patterns
- Range-based sharding works well for time-series data, while hash-based provides even distribution
- Cross-shard queries require scatter-gather patterns; minimize them for performance
- Rebalancing can be done online with dual-write patterns to avoid downtime
- Connection pooling must be managed per-shard to prevent resource exhaustion
- Monitoring both per-shard and cross-shard metrics is essential
Sharding is a significant architectural decision - consider using managed solutions like Vitess, CockroachDB, or Spanner if possible before implementing custom sharding.
Comments