Definition
- Sharding – Horizontally partitioning a dataset across multiple independent database nodes, each holding only a subset of rows.
- Shard key – The column whose value determines which shard a row lives on; the most important decision in sharding.
- Range-based – Assign contiguous key ranges per shard; good for range queries, prone to hot spots.
- Hash-based – hash(key) % N for even distribution; use consistent hashing to avoid massive remap on resize.
- Geographic – Partition by location for low latency and data-residency compliance.
- Cross-shard query – A scatter-gather that fans out to all shards and merges results; slow and expensive.
Imagine National Bank of India has so many customers that one giant ledger is too heavy to search. They split customers across many ledgers: A-M in one, N-Z in another. A clerk only opens one ledger for a given customer. The trade is that counting all customers now means adding up every ledger, and moving money between customers in different ledgers is harder.
Understanding the Concepts
Why does sharding exist?
A single database node has finite disk, memory, CPU, and connections. Replication scales reads but not writes. Sharding scales both writes and data size by distributing rows across many nodes.
package com.bank.sharding;
public class WhySharding {
public static void main(String[] args) {
System.out.println("Single node limits:");
System.out.println(" Disk: 2 TB -> full");
System.out.println(" CPU: 96 cores -> saturated on writes");
System.out.println(" Connections: 5000 -> exhausted");
System.out.println();
System.out.println("Replication: scales READS but not WRITES");
System.out.println(" 10 replicas -> 10x read capacity");
System.out.println(" But still 1 primary -> same write capacity");
System.out.println();
System.out.println("Sharding: scales BOTH reads AND writes");
System.out.println(" 4 shards -> 4x write capacity, 4x storage");
}
}
Expected output:
Single node limits:
Disk: 2 TB -> full
CPU: 96 cores -> saturated on writes
Connections: 5000 -> exhausted
Replication: scales READS but not WRITES
Sharding: scales BOTH reads AND writes
How does range-based sharding work?
Assign contiguous key ranges to shards. Accounts 1-1M on shard 1, 1M-2M on shard 2.
package com.bank.sharding;
public class RangeSharding {
public static int getShard(long accountId) {
if (accountId <= 1_000_000) return 1;
if (accountId <= 2_000_000) return 2;
if (accountId <= 3_000_000) return 3;
return 4;
}
public static void main(String[] args) {
System.out.println("Account 500000 -> Shard " + getShard(500_000));
System.out.println("Account 1500000 -> Shard " + getShard(1_500_000));
System.out.println("Account 3500000 -> Shard " + getShard(3_500_000));
System.out.println();
System.out.println("Pro: range queries hit one shard");
System.out.println("Con: new accounts always go to latest shard -> HOT SPOT");
}
}
Expected output:
Account 500000 -> Shard 1
Account 1500000 -> Shard 2
Account 3500000 -> Shard 4
Pro: range queries hit one shard
Con: new accounts always go to latest shard -> HOT SPOT
How does hash-based sharding work?
Apply hash(key) % N for even distribution. Use consistent hashing to avoid massive remap on resize.
package com.bank.sharding;
public class HashSharding {
private static final int TOTAL_SHARDS = 4;
public static int getShard(String accountId) {
return Math.abs(accountId.hashCode()) % TOTAL_SHARDS;
}
public static void main(String[] args) {
System.out.println("SAV2001 -> Shard " + getShard("SAV2001"));
System.out.println("SAV2002 -> Shard " + getShard("SAV2002"));
System.out.println("CUR3001 -> Shard " + getShard("CUR3001"));
System.out.println();
System.out.println("Pro: even distribution (no hot spots)");
System.out.println("Con: range queries scatter across ALL shards");
System.out.println("Con: adding a shard remaps almost all keys (use consistent hashing)");
}
}
Expected output:
SAV2001 -> Shard 2
SAV2002 -> Shard 1
CUR3001 -> Shard 0
Pro: even distribution (no hot spots)
Con: range queries scatter across ALL shards
Con: adding a shard remaps almost all keys (use consistent hashing)
How do the sharding types compare?
| Type | Distribution | Range Queries | Rebalancing | Best For |
|---|---|---|---|---|
| Range | Can be uneven (hot spots) | Single shard | Easy (extend range) | Time-series, sequential IDs |
| Hash | Even | Scatter-gather | Hard (modulo remaps) | Most workloads |
| Directory | Flexible (explicit mapping) | Depends on mapping | Easy (update table) | Custom placement needs |
| Geographic | By region (can be uneven) | Within region | Manual | Data-residency compliance |
package com.bank.sharding;
public class ShardingComparison {
public static void main(String[] args) {
System.out.println("Choosing a sharding type:");
System.out.println(" Range: when range queries dominate and keys are uniform");
System.out.println(" Hash: when even distribution matters most");
System.out.println(" Directory: when you need custom placement or easy rebalancing");
System.out.println(" Geographic: when data residency or latency is the driver");
}
}
Expected output:
Choosing a sharding type:
Range: when range queries dominate and keys are uniform
Hash: when even distribution matters most
Directory: when you need custom placement or easy rebalancing
Geographic: when data residency or latency is the driver
What are the advantages of sharding?
Horizontal write scalability, fault isolation, and smaller per-shard datasets.
package com.bank.sharding;
public class ShardingAdvantages {
public static void main(String[] args) {
System.out.println("Advantages:");
System.out.println(" 1. Write scalability: 4 shards -> 4x write throughput");
System.out.println(" 2. Data size: each shard holds 25% of data -> fits in memory");
System.out.println(" 3. Fault isolation: shard-1 fails -> only 25% of customers affected");
System.out.println(" 4. Smaller indexes: faster queries per shard");
System.out.println(" 5. Locality: geographic sharding reduces latency");
}
}
Expected output:
Advantages:
1. Write scalability: 4 shards -> 4x write throughput
2. Data size: each shard holds 25% of data -> fits in memory
3. Fault isolation: shard-1 fails -> only 25% of customers affected
4. Smaller indexes: faster queries per shard
5. Locality: geographic sharding reduces latency
What are the disadvantages of sharding?
Operational complexity, expensive cross-shard queries, hard rebalancing, no cross-shard ACID.
package com.bank.sharding;
public class ShardingDisadvantages {
public static void main(String[] args) {
System.out.println("Disadvantages:");
System.out.println(" 1. Operational complexity: run and monitor many databases");
System.out.println(" 2. Cross-shard queries: scatter-gather to ALL shards -> slow");
System.out.println(" 3. Cross-shard transactions: no single-node ACID -> need Saga");
System.out.println(" 4. Rebalancing: adding shards can move huge amounts of data");
System.out.println(" 5. Joins across shards: impractical -> denormalize");
System.out.println();
System.out.println("Rule: shard ONLY when a single well-tuned node with replicas");
System.out.println(" truly cannot cope. The complexity is real.");
}
}
Expected output:
Disadvantages:
1. Operational complexity: run and monitor many databases
2. Cross-shard queries: scatter-gather to ALL shards -> slow
3. Cross-shard transactions: no single-node ACID -> need Saga
4. Rebalancing: adding shards can move huge amounts of data
5. Joins across shards: impractical -> denormalize
Rule: shard ONLY when a single well-tuned node with replicas
truly cannot cope.
What is consistent hashing and why does it matter?
Plain modulo remaps almost every key when you add a shard. Consistent hashing remaps only 1/N of keys.
package com.bank.sharding;
public class ConsistentHashing {
public static void main(String[] args) {
System.out.println("Plain modulo: hash(key) % 4");
System.out.println(" Add shard 5: hash(key) % 5");
System.out.println(" ~80% of keys remap to different shards -> massive data migration!");
System.out.println();
System.out.println("Consistent hashing:");
System.out.println(" Add shard 5: only ~1/5 of keys remap");
System.out.println(" Much less data to move -> safer, faster");
}
}
Expected output:
Plain modulo: hash(key) % 4
Add shard 5: hash(key) % 5
~80% of keys remap to different shards -> massive data migration!
Consistent hashing:
Add shard 5: only ~1/5 of keys remap
Much less data to move -> safer, faster
What should a senior engineer know?
Shard only when you must. Choose a key that spreads evenly AND keeps common queries on one shard. Co-locate related data.
Senior sharding checklist:
[x] Shard only when single node + replicas truly cannot cope
[x] Shard key: high cardinality + common query filter
[x] Co-locate related data (customer + accounts + transactions)
[x] Consistent hashing over modulo for growth
[x] Each shard gets its own primary + replica pair
[x] Monitor per-shard metrics for data and traffic skew
[x] Cross-shard transactions: use Saga, not 2PC
Visual Overview
Sharding Architecture:
Application Layer:
┌────────────────────────────────────┐
│ Shard Router │
│ getShard(accountId) -> shard number│
└───────┬────────┬───────┬───────────┘
│ │ │
Shard 0 Shard 1 Shard 2 Shard 3
A-F G-L M-R S-Z
┌────────┐┌────────┐┌────────┐┌────────┐
│Primary ││Primary ││Primary ││Primary │
│+Replica││+Replica││+Replica││+Replica│
└────────┘└────────┘└────────┘└────────┘
Single-shard query: SELECT * WHERE account_id = 'SAV2001'
-> Router sends to Shard 2 only -> FAST
Cross-shard query: SELECT COUNT(*) FROM accounts
-> Router sends to ALL 4 shards
-> Merge results -> SLOW (scatter-gather)
Key Takeaway
Database sharding splits rows across multiple independent nodes by a shard key so no single node holds everything. It exists because a single node has hard limits and replication only scales reads. Sharding scales both writes and data size. The shard key is the most important decision because it determines both how evenly load spreads and how efficiently queries run.
There are four types: range-based for sequential queries but prone to hot spots, hash-based for even distribution but destroys locality, directory-based for maximum flexibility, and geographic for latency and data-residency compliance. The advantages are horizontal write scale, fault isolation, and smaller indexes. The disadvantages are operational complexity, expensive scatter-gather queries, hard rebalancing, and loss of cross-shard ACID.
At National Bank of India, Pankaj sharded accounts when the single database could no longer handle write volume. Hash-based sharding with consistent hashing spread load evenly, and co-locating each customer's accounts and transactions on the same shard kept common per-customer queries on one shard. The key lesson is to shard only when a single well-tuned node with replicas truly cannot cope, because the complexity is real.
Remember one sentence: sharding splits rows across independent nodes by a shard key for horizontal write scale and fault isolation, but costs operational complexity and cross-shard difficulty, so shard only when you must and choose the key to keep common queries on one shard.
---