Sharding divides a dataset into parts distributed among servers or storage groups. Each shard retains a portion of the dataset and handles the corresponding operations. The goal is to distribute capacity when a single unit can no longer sustain the required volume or load.
Dividing data also divides its locality. A query or rule that previously fit within one database may begin to require several participants. The partition key and routing must be chosen alongside those operations, rather than solely by record count.
Before splitting the database
First, identify the limit: CPU, storage, I/O, writes, reads, or contention on specific data. A query without an index and a database that has exhausted its write capacity require different responses.
Reviewing indexes, execution plans, result size, and repeated queries can reduce work without changing the distribution. Vertical scaling adds resources to one unit, within its limits. Replicas can distribute reads compatible with their lag, but do not automatically eliminate the primary’s write load.
Sharding fits when distributing data and operations can overcome the identified bottleneck and its operational cost is acceptable. Measurements must include p95 and p99 latency, growth, concurrency, and headroom during failures. Adding nodes does not guarantee proportional improvement if work remains concentrated on a single entity.
Sizing before migrating
The decision rests on measured sustainable capacity, expected growth, and the time needed to migrate. Headroom should be reserved before reaching an absolute limit. Locality or workload isolation can also justify earlier separation, with explicit costs.
As a hypothetical example, net growth of new data of 20 GB per day, retained for 365 days, produces 7,300 GB = 7.3 TB of raw data. Decimal units are used: 1 TB equals 1,000 GB. Assuming an additional 40% for indexes and overhead adds 2.92 TB and brings the dataset to 10.22 TB per copy. One primary and two full replicas would total 30.66 TB of aggregate physical storage for those copies. Logs, backups, and temporary migration space are budgeted separately; those percentages are assumptions, not universal properties.
The calculation sizes storage. Reads and writes must be tested with their real mix of queries, indexes, bytes modified, concurrency, replication, and durability. The load that sustains the target latency and headroom under failures is measured; dividing an average request rate among nodes does not demonstrate that capacity.
Figures such as hundreds of TB or thousands of writes per second need those conditions to be interpreted. Measured size, rate, and cost determine whether optimizing, scaling one unit, or distributing work meets the requirement.
Choosing the unit and key
The partition key determines which data is grouped together. It can be tenant_id, user_id, or another unit aligned with important queries and transactions. The choice should favor colocation of data read or modified together.
For example, grouping by tenant_id can keep a tenant’s users and orders in the same partition. A query that knows the tenant can be directed to its owner. Searching orders by a global criterion, without a tenant, may require fanout or an additional index that must also be maintained.
High cardinality provides many units to distribute, but does not guarantee balance. One tenant may have more data, requests, or expensive writes than thousands of others. Size and work per key must be observed alongside key count.
Colocating related data allows certain decisions to remain within a local transaction. Constraints also have a scope: checking uniqueness within one shard does not demonstrate global uniqueness. If the rule crosses tenants or partitions, it needs another authority or explicit coordination. The key is chosen with those invariants and expected growth in mind.
Partitioning and routing
Logical grouping and physical placement are related decisions. Hashes or ranges can define groups; a directory can assign them to owners. These techniques can be combined.
| Mechanism | What it addresses | Main trade-off |
|---|---|---|
| Key hashing | Groups keys through a partitioning function. | Disperses keys, but can turn range queries into fanout. |
| Ranges | Groups intervals of an ordered key. | Helps certain queries; unequal ranges or the newest end can concentrate load. |
| Logical partition directory | Maps a partition to its physical owner. | Allows groups to move, but requires maintaining and updating routes. |
With monotonic keys, such as increasing identifiers, new writes can accumulate in the last range. With hashing, a highly active user still concentrates work if its entire entity is assigned to one group.
Consistent hashing can reduce assignment movement when topology changes compared with a scheme that remaps much of the dataset. It does not eliminate migration, write transfer, or route coordination. Virtual nodes allow more assignment units to be distributed; they do not guarantee uniform bytes, requests, or cost.
--- config: look: handDrawn --- flowchart TD app["Application<br/>Query + key"] --> router["Router<br/>Key to logical partition"] router -.->|"Resolve location"| directory[("Versioned directory<br/>Partition to owner")] directory -.->|"Owner + version"| router router -->|"Group A"| shardA[("Shard 1<br/>Assigned partitions")] router -->|"Group B"| shardB[("Shard 2<br/>Assigned partitions")] router -->|"Group C"| shardC[("Shard 3<br/>Assigned partitions")]
The diagram represents a query with a known key and three possible owners. The directory provides placement metadata; the router directs the operation to the corresponding shard. Routes can be cached, but must be refreshable and able to detect old versions during redistribution.
Hotspots and large entities
A user with a massive audience or a large tenant can saturate its shard. A dedicated shard isolates that load and allows it to be assigned its own resources, but does not provide unlimited capacity or automatically split an entity.
A compound key such as (user_id, bucket) can distribute its data if the bucket actually participates in partitioning and separates units of work. Adding it without changing the assignment redistributes nothing. Splitting also introduces costs: assembling history may touch several shards, and maintaining order or atomicity across buckets requires coordination.
If a rule serializes all updates to the same balance or record, distributing other data does not remove that contention. The design must decide which operations are independent and which need to retain a common authority.
For repeated reads, SD: Caching describes reusable copies and the limits of hot keys. The cache can offload queries compatible with its freshness; it does not make distributed writes atomic or eliminate the cost of rebuilding results. Isolating load and limiting admission remain capacity decisions.
Cross-shard queries and transactions
A query without a key that can locate the data may be sent to several shards. That fanout adds requests, work, and a merge phase. Latency may depend on the slowest participant; a partial failure requires choosing between rejection, retry, or an incomplete result accepted by the product.
Global sorting and pagination require a stable criterion and a way to continue across participants. If data changes between pages, omissions or repetitions may occur. When a coherent view of the dataset is needed, the design must define a snapshot or read contract that coordinates that consistency.
Denormalization, global indexes, and projections can avoid some fanout for specific queries. They add storage, maintenance writes, and possible staleness windows. They do not make every query cheap or provide atomicity by themselves.
For a strong invariant, data should be colocated when possible, or a system supporting the necessary distributed transactions should be used, with its guarantees and costs reviewed.
A transfer between Bob and Alice
Suppose Bob sends $5 to Alice, in the same currency and without fees. In a single authoritative database, a transaction can debit $5 from Bob, credit $5 to Alice, and record ledger entries durably. Atomicity commits those changes together or aborts them. Isolation, locks, or update conditions must protect the available balance against concurrent transfers and prevent overdrafts. The transfer identifier also allows a retry to be recognized.
If the accounts reside on different shards, that local boundary no longer covers both changes. A system with compatible participants can use 2PC to prepare its state durably and agree on a commit or abort decision. It adds coordination and latency; a prepared participant may retain resources while recovering an uncertain decision. A coordinator failure does not mean the entire database becomes unusable. Isolation guarantees still depend on the implementation.
A Saga requires a different contract. It can commit a local debit together with a pending transfer, then credit Alice, and finally close the flow. During that interval, the $5 must be represented as funds in transit, and the operation has observable state. It fits only if the business accepts those intermediate states and their recovery; it does not provide global atomicity.
Each step retains transfer_id, its result, and the local effect with atomic coordination to tolerate retries. If the credit to Alice times out, it may have occurred. Before deciding on compensation, its result is recovered or queried, or the operation is retried under the idempotency contract. The decision must be coordinated with late credits: observing that a credit has not yet appeared is insufficient if it can still complete.
Compensation may require a new ledger entry returning funds to Bob, without deleting the original debit. Doing so automatically after an uncertain result could return money to him while Alice has already received it. Compensation can also fail and remain pending reconciliation. Recovery needs durable records, retries, and failure resolution; it does not guarantee automatic convergence.
If the rule requires both sides to change atomically, an authority or ledger with the appropriate transaction boundary is retained, or distributed guarantees satisfying it are chosen. Compensation does not replace that rule. SD: 2PC vs Saga explains the distinction between atomicity, isolation, and recovery.
Migration and operations
Moving a partition requires copying data and resolving concurrent writes. While the source retains authority, changes produced during the copy must be transferred through a recoverable mechanism, with ordering or versions and idempotency. Verifying only the initial copy does not demonstrate that the destination is up to date.
The cutover establishes when ownership changes and which route version identifies it. The source must stop accepting operations as the previous owner or apply the intended redirection. Clients with old routes must refresh them or receive a response that allows them to be resolved. The design needs to prevent two owners from accepting incompatible writes.
The transfer must be verified before the change and retain a recovery path. It may require bounding or pausing writes during part of the process. After cutover, returning to an old copy without reconciling new writes can lose data. Redistribution has network, storage, and capacity costs even with consistent hashing.
Sharding and replication are different. The former distributes portions; the latter retains copies of a portion. Each shard needs its high availability, replication, backup, and restoration strategy. Having more shards does not automatically provide those guarantees and adds units that can fail.
What to verify
Tests must concentrate traffic on one key, fill a recent range, and compare local queries with fanout. They must also move a partition while writes arrive, introduce old routes, and stop participants before and after cutover. Recovery must explain which owner has authority and which operations were committed or remain pending.
Useful metrics are observed per shard and partition: bytes, growth, CPU, I/O, requests, query cost, and p95 and p99 latency. Concentration by key, replication lag, pending migration work, and operations requiring multiple participants also matter. A balanced average can hide a saturated shard.
In an interview, the design must justify the key, operations that remain local, and recovery when placement changes. Capacity gained is assessed alongside the queries, business rules, and operational tasks that now cross that boundary.