Definition
- Shard key – The field whose value determines which shard a row lives on; once data is distributed, changing it requires a painful re-shard.
- Even distribution – Data and request traffic spread roughly equally across all shards so no node is overloaded.
- Query locality – Common queries filter on the shard key and hit a single shard instead of scatter-gathering to all.
- Hot shard – One node receiving disproportionate data or traffic that saturates its resources and caps the cluster.
- Transaction boundary – Single-shard operations keep ACID; cross-shard operations need distributed coordination.
Imagine National Bank of India assigns each customer to a counter based on some rule. A good rule spreads customers evenly so every counter is equally busy, and keeps each customer's related business at one counter. A bad rule, like sending everyone who opened an account today to the same counter, piles a huge crowd at that one counter. That overloaded counter is the hot shard: it caps how fast the whole branch can serve customers.
Understanding the Concepts
Why is choosing a shard key so important?
The decision is effectively permanent. Once data is distributed and the application routes by that key, changing it means re-sharding the entire dataset.
package com.bank.sharding;
public class ShardKeyImportance {
public static void main(String[] args) {
System.out.println("Shard key: the MOST important decision in sharding");
System.out.println(" Once chosen, changing it = re-shard entire dataset");
System.out.println(" Re-sharding = move terabytes of data while serving traffic");
System.out.println(" One of the most painful operations in distributed databases");
System.out.println();
System.out.println("Two properties to balance:");
System.out.println(" 1. Even distribution (no hot shard)");
System.out.println(" 2. Query locality (common queries hit one shard)");
}
}
Expected output:
Shard key: the MOST important decision in sharding
Once chosen, changing it = re-shard entire dataset
Re-sharding = move terabytes of data while serving traffic
One of the most painful operations in distributed databases
Two properties to balance:
1. Even distribution (no hot shard)
2. Query locality (common queries hit one shard)
What is a hot shard and why does it cap scalability?
A sharded cluster is only as fast as its busiest shard. If one shard holds or serves far more than its share, it saturates while others sit idle.
package com.bank.sharding;
public class HotShard {
public static void main(String[] args) {
System.out.println("Hot shard scenario:");
System.out.println(" 4 shards, each should handle ~25% of traffic");
System.out.println(" Bad key: shard-0 gets 70%, shards 1-3 get 10% each");
System.out.println(" Shard-0 saturates at 5000 req/sec");
System.out.println(" Total system capped at ~7000 req/sec");
System.out.println(" Adding more shards does NOT help");
System.out.println(" The hot shard IS the bottleneck");
System.out.println();
System.out.println(" Fix: re-shard with a better key (painful but necessary)");
}
}
Expected output:
Hot shard scenario:
4 shards, each should handle ~25% of traffic
Bad key: shard-0 gets 70%, shards 1-3 get 10% each
Shard-0 saturates at 5000 req/sec
Total system capped at ~7000 req/sec
Adding more shards does NOT help
The hot shard IS the bottleneck
What causes a hot shard?
Three common causes: low-cardinality keys, monotonic keys, and popularity-correlated keys.
package com.bank.sharding;
public class HotShardCauses {
public static void main(String[] args) {
System.out.println("Cause 1: Low cardinality");
System.out.println(" Key: country (India/USA/UK)");
System.out.println(" 90% of customers are in India -> one shard gets 90%");
System.out.println();
System.out.println("Cause 2: Monotonic key");
System.out.println(" Key: auto-increment ID with range sharding");
System.out.println(" All new accounts go to the latest shard -> always hot");
System.out.println();
System.out.println("Cause 3: Popularity-correlated");
System.out.println(" Key: merchant_id");
System.out.println(" Amazon's merchant_id on one shard -> massive traffic");
}
}
Expected output:
Cause 1: Low cardinality
Key: country (India/USA/UK)
90% of customers are in India -> one shard gets 90%
Cause 2: Monotonic key
Key: auto-increment ID with range sharding
All new accounts go to the latest shard -> always hot
Cause 3: Popularity-correlated
Key: merchant_id
Amazon's merchant_id on one shard -> massive traffic
Why must the key provide query locality?
A query without the shard key cannot be routed to one node. It scatter-gathers to every shard and merges.
package com.bank.sharding;
public class QueryLocality {
public static void main(String[] args) {
System.out.println("Good locality (shard key = customer_id):");
System.out.println(" SELECT * FROM transactions WHERE customer_id = 'SAV2001'");
System.out.println(" -> Router sends to ONE shard -> fast");
System.out.println();
System.out.println("Bad locality (shard key = random hash):");
System.out.println(" SELECT * FROM transactions WHERE customer_id = 'SAV2001'");
System.out.println(" -> customer_id is NOT the shard key");
System.out.println(" -> Router sends to ALL 4 shards -> scatter-gather -> slow");
}
}
Expected output:
Good locality (shard key = customer_id):
SELECT * FROM transactions WHERE customer_id = 'SAV2001'
-> Router sends to ONE shard -> fast
Bad locality (shard key = random hash):
SELECT * FROM transactions WHERE customer_id = 'SAV2001'
-> Router sends to ALL 4 shards -> scatter-gather -> slow
Why is customer_id usually the best key for banking?
High cardinality for even spread, plus natural filter for per-customer queries.
package com.bank.sharding;
public class CustomerIdAsKey {
public static void main(String[] args) {
System.out.println("customer_id as shard key:");
System.out.println(" High cardinality: millions of unique customers");
System.out.println(" -> Hashing spreads evenly, no hot shard");
System.out.println();
System.out.println(" Common queries filter on customer_id:");
System.out.println(" 'Show Rajesh's accounts' -> one shard");
System.out.println(" 'Show Rajesh's transactions' -> one shard");
System.out.println(" 'Show Rajesh's statements' -> one shard");
System.out.println();
System.out.println(" Co-located data: accounts + transactions + statements");
System.out.println(" All on the same shard for one customer");
}
}
Expected output:
customer_id as shard key:
High cardinality: millions of unique customers
-> Hashing spreads evenly, no hot shard
Common queries filter on customer_id:
'Show Rajesh's accounts' -> one shard
'Show Rajesh's transactions' -> one shard
'Show Rajesh's statements' -> one shard
Co-located data: accounts + transactions + statements
Why is the shard key also the transaction boundary?
Single-shard operations keep ACID. Cross-shard operations need distributed coordination (Saga).
package com.bank.sharding;
public class TransactionBoundary {
public static void main(String[] args) {
System.out.println("Shard key = transaction boundary:");
System.out.println();
System.out.println(" Rajesh's accounts all on Shard 2:");
System.out.println(" Transfer between Rajesh's savings and current");
System.out.println(" -> Single shard -> ACID guaranteed");
System.out.println();
System.out.println(" Transfer from Rajesh (Shard 2) to Priya (Shard 3):");
System.out.println(" -> Cross-shard -> NO single-node ACID");
System.out.println(" -> Need Saga with compensating transactions");
}
}
Expected output:
Shard key = transaction boundary:
Rajesh's accounts all on Shard 2:
Transfer between Rajesh's savings and current
-> Single shard -> ACID guaranteed
Transfer from Rajesh (Shard 2) to Priya (Shard 3):
-> Cross-shard -> NO single-node ACID
-> Need Saga with compensating transactions
What goes wrong with bad keys? Real examples.
package com.bank.sharding;
public class BadKeyExamples {
public static void main(String[] args) {
System.out.println("Bad key 1: account_opening_date (monotonic)");
System.out.println(" All new accounts -> latest shard -> HOT");
System.out.println();
System.out.println("Bad key 2: region (low cardinality)");
System.out.println(" 80% of customers in Maharashtra -> one shard gets 80%");
System.out.println();
System.out.println("Good key: customer_id (high cardinality + common filter)");
System.out.println(" Even spread under hashing");
System.out.println(" Per-customer queries stay on one shard");
}
}
Expected output:
Bad key 1: account_opening_date (monotonic)
All new accounts -> latest shard -> HOT
Bad key 2: region (low cardinality)
80% of customers in Maharashtra -> one shard gets 80%
Good key: customer_id (high cardinality + common filter)
Even spread under hashing
Per-customer queries stay on one shard
What should a senior engineer know?
Hash monotonic keys, co-locate related entities, handle naturally hot entities specially, and monitor per-shard skew.
Senior guidelines:
[x] Choose key: high cardinality + common query filter
[x] Hash monotonic keys to kill write hot spots
[x] Co-locate: customer + accounts + transactions on same shard
[x] Handle hot entities (e.g. corporate mega-account) specially
[x] Monitor per-shard: data size, CPU, connections, traffic
[x] Shard key = transaction boundary: plan cross-shard flows
[x] Prefer consistent hashing for growth
[x] The fix for a bad key is a painful re-shard
Visual Overview
Shard Key Decision: Good Key (customer_id): ┌─────────────────────────────┐ │ customer_id = SAV2001 │ -> hash -> Shard 2 │ All of Rajesh's data: │ │ accounts, transactions, │ │ statements │ │ Common query: WHERE cust=X │ -> hits ONE shard └─────────────────────────────┘ Bad Key (country): ┌─────────────────────────────┐ │ country = 'India' │ -> Shard 0 │ 90% of all customers! │ -> HOT SHARD │ Shard 0 saturated. │ │ Shards 1,2,3 nearly empty. │ └─────────────────────────────┘ Bad Key (auto-increment ID with range): ┌─────────────────────────────┐ │ ID 1-1M -> Shard 0 (cold) │ │ ID 1M-2M -> Shard 1 (cold) │ │ ID 2M-3M -> Shard 2 (cold) │ │ ID 3M-4M -> Shard 3 (HOT) │ <- all new writes here └─────────────────────────────┘
Key Takeaway
The shard key is the field that decides which shard a row lives on, and it is the single most consequential decision in sharding because it is effectively permanent. A good shard key must do two things at once: distribute data and traffic evenly so no shard becomes a bottleneck, and keep common queries localized to one shard so they do not scatter-gather.
A hot shard is one node receiving disproportionate load. It caps the entire cluster because the system is only as fast as its busiest shard, and adding more shards does not help. Hot shards come from low-cardinality keys (country), monotonic keys (auto-increment ID), or popularity-correlated keys (celebrity merchant). Poor locality scatters even simple queries to every shard.
At National Bank of India, Pankaj first sharded by opening date (monotonic), creating a hot shard for new accounts. They tried region, but most customers were in one state. Switching to customer_id fixed both: it spread evenly under hashing and kept per-customer queries on one shard. The key is also the transaction boundary: single-shard operations keep ACID while cross-shard ones need a Saga.
Remember one sentence: a good shard key must spread data evenly so no node becomes a hot shard AND keep common queries on one shard, and in banking customer_id usually wins because a bad key creates a bottleneck that no amount of extra shards can fix.
---