Replication, Partitioning, and Sharding Questions
Scaling and distributing data across nodes: primary-replica and multi-primary replication, read-replica scaling, horizontal partitioning, and sharding strategies with their key-selection and rebalancing challenges. Covers replication lag, failover and split-brain handling, cross-shard operations such as joins, distributed transactions, and global secondary indexes, and the operational cost of a partitioned topology. Key to designing databases that scale horizontally.
What is consistent hashing and how does it reduce data movement when the number of shards changes? Explain the role of virtual nodes and how they improve load distribution on the hash ring.
Sample Answer
Direct answer
Consistent hashing maps both data keys and shard nodes onto the same circular numeric space (a "ring"), so a key belongs to whichever node's position is the next one clockwise from the key's hash. Because a node's ring position is independent of every other node, adding or removing one node only reassigns the keys in the arc next to it, roughly 1/N of all keys for N nodes, instead of remapping almost everything the way hash(key) mod N does when N changes. Virtual nodes (vnodes) give each physical shard many scattered positions on the ring instead of one, which smooths out the uneven arc sizes a handful of random points would otherwise produce and lets you hand a bigger shard proportionally more of the ring.
Structured elaboration
The ring mechanism
- Hash both shard identifiers and keys with the same hash function onto a fixed range, for example 0 to 2^32-1 ("the ring"), and treat that range as wrapping around like a clock face.
- Each physical shard is assigned one or more positions ("tokens") on the ring.
- To find a key's owner: hash the key, then walk clockwise from that point to the first token; that token's shard owns the key.
Why this bounds data movement when the shard count changes
The naive scheme owner(key) = hash(key) mod N recomputes every key's owner from scratch whenever N changes, because the modulus itself changed; in practice that remaps close to all keys (demonstrated below: 85.5% for a 6-to-7 shard change). Consistent hashing only moves the keys whose ring position falls inside the arc that changes ownership. Insert a 7th node's token and it slices a piece off of whichever single node's arc it lands inside; every other node's arc is untouched. The maximum fraction remapped is therefore bounded by roughly the new node's share of the ring (about 1/7, 14.3%, for 6 to 7 shards), and only the new node gains keys; the six existing shards never trade keys with each other.
The role of virtual nodes
A physical shard placed at a single random ring position owns one contiguous arc, and with only a handful of shards those arc lengths vary a lot by chance, so some shards end up owning far more of the ring than others even though the expected share is 1/N for everyone. Giving each shard many small, scattered arcs instead of one large one buys three things:
- Load smoothing: with many small arcs per shard, the total arc length each shard owns converges toward its fair share as vnode count grows (demonstrated below: standard deviation drops from about 50% of the mean at 1 vnode/shard to about 4% at 100 vnodes/shard).
- Uneven capacity, on request: assigning a bigger shard more vnodes than a smaller one gives it a proportionally larger slice of the ring without changing the hashing algorithm at all: a node with twice the capacity of its peers simply gets twice as many tokens, so its expected share of ring traffic scales with its vnode count.
- Even rebalancing on add or remove: because a shard's ownership is spread across many small arcs instead of one, when it joins or leaves, the affected keys come from, or go to, many other shards roughly evenly instead of dumping the whole burden on whichever one shard happened to be its ring neighbor.
Choosing a vnode count
More vnodes improve balance but are not free:
- Metadata and coordination cost: every vnode is an entry the routing layer, or every node in a gossip-based system, must store and keep synchronized; at very high counts (hundreds of thousands across a cluster) that metadata and its propagation traffic become a real operational tax.
- Diminishing returns: balance improves fastest at low counts and flattens out, because it is a statistical variance-reduction effect, not a fixed formula, so moving from 1 to 32 vnodes buys far more balance than moving from 128 to 256 does.
- A concrete, checkable data point: Apache Cassandra (a widely used distributed, wide-column NoSQL database, an example of exactly the kind of node ring described above) shipped for years with 256 tokens (vnodes) per node by default, to guarantee smooth balance through brute-force randomness alone. Cassandra 4.0 lowered the shipped default to 16, paired with a replication-factor-aware token allocation algorithm (
allocate_tokens_for_local_replication_factor) that deliberately places each new node's tokens for balance, rather than relying on a large random count to average out skew. The project's own public debate over the new default (proposals ranged from 4 to 16 tokens) weighed availability and repair blast-radius against small-cluster balance, and settled on 16 as the practical compromise (Apache Cassandra dev mailing list, "num_tokens default in Cassandra 4.0" thread). - Rule of thumb: pick a base vnode count that keeps the worst-case imbalance,
max shard share / mean shard share, under whatever your operations team can tolerate at the smallest cluster size you expect to run, and prefer placement-aware allocation over pure randomness once the vnode count is high enough that its metadata cost is noticeable.
Worked example
This is a full, runnable simulation (Python standard library only, deterministic hashing via hashlib.md5, no randomness) comparing naive mod-N reassignment to consistent hashing at 1 and 100 vnodes/shard when a 6-shard ring grows to 7 shards, then measuring how vnode count affects load balance across a fixed 6-shard ring.
import hashlib, statistics
def h(key):
return int(hashlib.md5(key.encode()).hexdigest(), 16)
RING_SIZE = 2**32
def ring_hash(key):
return h(key) % RING_SIZE
NUM_KEYS = 20000
keys = [f"key-{i}" for i in range(NUM_KEYS)]
def naive_assign(key, n):
return h(key) % n
moved_naive = sum(1 for k in keys if naive_assign(k, 6) != naive_assign(k, 7))
print(f"naive hash-mod-N: nodes 6->7, keys remapped = {moved_naive}/{NUM_KEYS} = {moved_naive/NUM_KEYS:.1%}")
def build_ring(node_names, vnodes_per_node):
ring = {}
for node in node_names:
for v in range(vnodes_per_node):
token = ring_hash(f"{node}#{v}")
ring[token] = node
return sorted(ring.items())
def lookup(sorted_ring, key):
tok = ring_hash(key)
lo, hi = 0, len(sorted_ring)
while lo < hi:
mid = (lo + hi) // 2
if sorted_ring[mid][0] < tok:
lo = mid + 1
else:
hi = mid
if lo == len(sorted_ring):
lo = 0
return sorted_ring[lo][1]
nodes6 = [f"shard-{i}" for i in range(6)]
nodes7 = nodes6 + ["shard-6"]
for vnodes in (1, 100):
ring6, ring7 = build_ring(nodes6, vnodes), build_ring(nodes7, vnodes)
before = {k: lookup(ring6, k) for k in keys}
after = {k: lookup(ring7, k) for k in keys}
moved = sum(1 for k in keys if before[k] != after[k])
print(f"consistent hashing, {vnodes} vnode(s)/node: nodes 6->7, keys remapped = {moved}/{NUM_KEYS} = {moved/NUM_KEYS:.1%}")
print()
for vnodes in (1, 8, 32, 100, 256):
ring = build_ring(nodes6, vnodes)
counts = {n: 0 for n in nodes6}
for k in keys:
counts[lookup(ring, k)] += 1
vals = list(counts.values())
mean, stdev = statistics.mean(vals), statistics.pstdev(vals)
print(f"vnodes/node={vnodes:>3}: per-shard key counts min={min(vals)} max={max(vals)} mean={mean:.0f} "
f"stdev={stdev:.0f} ({stdev/mean:.1%} of mean) max/mean={max(vals)/mean:.2f}x")
Output:
naive hash-mod-N: nodes 6->7, keys remapped = 17098/20000 = 85.5%
consistent hashing, 1 vnode(s)/node: nodes 6->7, keys remapped = 1924/20000 = 9.6%
consistent hashing, 100 vnode(s)/node: nodes 6->7, keys remapped = 2595/20000 = 13.0%
vnodes/node= 1: per-shard key counts min=1028 max=5885 mean=3333 stdev=1663 (49.9% of mean) max/mean=1.77x
vnodes/node= 8: per-shard key counts min=2170 max=4218 mean=3333 stdev=713 (21.4% of mean) max/mean=1.27x
vnodes/node= 32: per-shard key counts min=2634 max=4290 mean=3333 stdev=538 (16.1% of mean) max/mean=1.29x
vnodes/node=100: per-shard key counts min=3187 max=3574 mean=3333 stdev=136 (4.1% of mean) max/mean=1.07x
vnodes/node=256: per-shard key counts min=3140 max=3696 mean=3333 stdev=213 (6.4% of mean) max/mean=1.11x
Two things to read off this: naive mod-N remaps 85.5% of keys on a one-node change, versus 9.6 to 13.0% for consistent hashing (both close to the roughly 14.3% theoretical minimum of 1/7), confirming the data-movement bound. Separately, going from 1 to 100 vnodes per node shrinks the standard deviation of per-shard load from about half the mean to about 4% of it, confirming the load-smoothing effect; note the 1-vnode run still lands near the 14.3% theoretical remap fraction on average even though its per-shard load is far less even, because the two effects (data moved on a topology change vs. steady-state balance) are related but distinct.
Trade-offs and pitfalls
- Consistent hashing alone still treats every physical shard as equally capable; without weighting via vnode count, a bigger box receives no more traffic than a small one. Fixing that just means scaling vnode count with capacity, more tokens for a bigger box, fewer for a smaller one, so the ring hands out load in proportion to what each shard can actually handle.
- Drawback worth naming directly: the ring's vnode-to-shard mapping is itself state that must be distributed and kept in sync, whether via gossip or a directory service, and a stale copy on any client routes requests to the wrong shard until refreshed. This coordination cost grows with vnode count.
- If the hash function used to place tokens is not close to uniform, the "roughly 1/N" bound degrades; always use a well-distributed hash such as MD5, MurmurHash, or a SHA variant, never something like a raw modulo of a monotonically increasing ID as the ring hash.
- Changing the hash function, or the ring's overall size, on a live cluster is equivalent to changing the entire assignment function and remaps close to 100% of keys; ring parameters are effectively part of the schema, not something to casually redeploy.
Explain database federation as an alternative to sharding: querying across multiple independently-owned databases without redistributing their data. Describe one scenario where federation is preferable to sharding, and one where federation introduces unacceptable complexity.
Sample Answer
Direct answer
Database federation queries across multiple databases that each remain independently owned, operated, and never redistributed, typically via a federation layer that fans a query out to the relevant source databases and stitches the results together. Sharding, by contrast, deliberately redistributes ONE logical dataset across multiple nodes that a single system controls end to end. The difference that matters in practice: sharding is something you design and own, choosing the shard key and rebalancing strategy yourself; federation is something you do TO data you often do not own or cannot move, because it already lives in separately operated systems (a different team's service, an acquired company's legacy database, a third-party system you only have read access to).
When federation is the right call
A company acquires another company whose customer data lives in its own separate, actively-used production database, still owned and operated by that team, with its own release cadence and its own operational constraints. Migrating that data into a shared, sharded schema is a large, risky project with no urgent deadline forcing it. Federation lets a unified "search across all customers" feature query both databases live and merge the results, without either team giving up ownership of their own system or taking on a risky migration before there is a clear business reason to. This is the right trade when the data's owners, schemas, and operational lifecycles are genuinely going to stay independent for the foreseeable future, and the query patterns across them are occasional cross-source lookups rather than the bulk of the workload.
When federation introduces unacceptable complexity
A query that needs to join large amounts of data across two of the federated sources (a report needing every acquired company's transactions joined against the main company's customer table, filtered and sorted across the combined set) turns into pulling large result sets out of both systems over the network and joining them in application code or a federation middleware layer, since neither underlying database can push down a join against data it does not have. This scales badly and gets slower exactly as both datasets grow, with no way to add an index that spans both sides of the join. At that point, either the query pattern needs to change (accept eventually-consistent denormalized copies, replicated via change-data-capture (CDC: a technique that watches a source database's transaction log for every row-level insert, update, and delete, then streams those changes out as events, instead of querying the source live for each one) into one place purpose-built for the cross-source query, rather than querying live), or the underlying decision to keep the data federated instead of migrated needs to be revisited, since federation was never designed to make heavy cross-source joins fast.
Trade-offs and pitfalls
- Federation trades query performance and consistency for organizational and operational independence. It is the right tool when that independence is the actual constraint (you cannot or should not move the data), not a substitute for sharding when you simply have not gotten around to a proper data architecture.
- A federation layer becomes a hidden single point of failure and a hidden performance bottleneck if every cross-source query has to flow through it, even though no single underlying database is overloaded.
- Schema drift across the federated sources is a real ongoing cost: each source can and will change its own schema independently, on its own schedule, and the federation layer has to keep working across those changes without any coordinated migration process to keep everything in sync.
Develop a cost model to compare two choices for scaling a sharded datastore: adding more smaller instances versus upgrading to fewer larger instances. Include hardware, licensing, operational overhead, failover impact, and migration complexity in your model and describe how you'd apply it to a decision.
Sample Answer
Direct answer
Hold total capacity fixed and compare the two topologies on the dimensions that actually differ, since raw compute cost is close to a wash: most cloud instance families price roughly linearly with capacity within the family, so 5x the capacity costs close to 5x the hourly rate either way. The real trade-off is fixed per-instance costs (operational overhead, and for per-core-licensed engines, licensing) working in favor of fewer, larger instances, against failover blast radius and migration granularity working in favor of many, smaller ones. Build the model, then apply the workload's actual constraints (how much does an operator-hour cost, and how much write-capacity loss during a single-instance failure is tolerable) to pick a side, rather than defaulting to whichever is cheaper on the compute line alone.
The model
Inputs: a target sustained capacity (queries per second, QPS), a chosen instance size for each topology with its capacity and hourly price, a fixed monthly operational-overhead cost per instance (patch windows, monitoring and alerting setup, on-call cognitive load, backup scheduling, amortized as labor cost), and a redundancy factor (replicas per primary for high availability). From these, compute: total instance count for each topology, total monthly compute cost, total monthly operational cost, and blast radius (the fraction of total write capacity lost if exactly one primary instance fails).
"""Cost model: many-small-instances vs fewer-large-instances for a sharded datastore
holding a FIXED total capacity target. All unit costs below are stated, illustrative
assumptions (labeled as such), not live vendor pricing -- the model's structure and
the resulting trade-off is the point, not the specific dollar figures."""
TARGET_QPS = 400_000 # sustained peak the cluster must serve
REPLICAS_PER_SHARD = 1 # +1 synchronous replica per primary for HA (synchronous: must confirm a write before it counts as committed), both configs
# -- illustrative unit economics (compute pricing assumed LINEAR in capacity within
# a family: 5x the capacity costs 5x the hourly price -- realistic for most
# compute-optimized cloud instance families, stated as a simplifying assumption) --
SMALL_CAPACITY_QPS = 20_000
SMALL_PRICE_PER_HR = 0.68
LARGE_CAPACITY_QPS = 100_000 # 5x the small instance's capacity
LARGE_PRICE_PER_HR = SMALL_PRICE_PER_HR * (LARGE_CAPACITY_QPS / SMALL_CAPACITY_QPS)
FIXED_OPS_OVERHEAD_PER_INSTANCE_MONTHLY = 150 # patch windows, monitoring/alerting
# setup, on-call load, backup scheduling
# -- amortized labor, illustrative
HOURS_PER_MONTH = 730
def cluster_plan(capacity_per_primary, price_per_hr_per_instance):
num_primaries = -(-TARGET_QPS // capacity_per_primary) # ceil
total_instances = num_primaries * (1 + REPLICAS_PER_SHARD)
compute_cost = total_instances * price_per_hr_per_instance * HOURS_PER_MONTH
ops_cost = total_instances * FIXED_OPS_OVERHEAD_PER_INSTANCE_MONTHLY
blast_radius_pct = 1 / num_primaries * 100 # % of write capacity lost if 1 primary fails
return {
"num_primaries": num_primaries,
"total_instances": total_instances,
"compute_cost_monthly": compute_cost,
"ops_cost_monthly": ops_cost,
"total_monthly": compute_cost + ops_cost,
"blast_radius_pct": blast_radius_pct,
}
small = cluster_plan(SMALL_CAPACITY_QPS, SMALL_PRICE_PER_HR)
large = cluster_plan(LARGE_CAPACITY_QPS, LARGE_PRICE_PER_HR)
print(f"target: {TARGET_QPS:,} QPS sustained, {REPLICAS_PER_SHARD} replica/primary\n")
for name, plan in (("many small instances", small), ("fewer large instances", large)):
print(f"-- {name} ({plan['num_primaries']} primaries, {plan['total_instances']} instances total) --")
print(f" compute: ${plan['compute_cost_monthly']:,.0f}/mo")
print(f" ops overhead: ${plan['ops_cost_monthly']:,.0f}/mo")
print(f" TOTAL: ${plan['total_monthly']:,.0f}/mo")
print(f" blast radius if one primary fails: {plan['blast_radius_pct']:.1f}% of write capacity")
print()
savings = small["total_monthly"] - large["total_monthly"]
savings_pct = savings / small["total_monthly"] * 100
print(f"fewer-large saves ${savings:,.0f}/mo ({savings_pct:.1f}%) at this target, driven entirely by "
f"the {small['total_instances'] - large['total_instances']} fewer instances' worth of fixed ops "
f"overhead -- compute cost itself is ~equal by construction (linear pricing assumption).")
print()
# -- resharding friction: how many individual shard-migration operations to move from
# the current topology to 2x capacity, one direction each way --
def split_steps(num_primaries):
# each split takes 1 shard -> 2 shards; going from N to 2N needs N split operations
return num_primaries
print("-- resharding friction: operations needed to DOUBLE total capacity --")
print(f" many-small ({small['num_primaries']} primaries): {split_steps(small['num_primaries'])} "
f"individual shard-split operations, each moving ~1/{small['num_primaries']} of the data")
print(f" fewer-large ({large['num_primaries']} primaries): {split_steps(large['num_primaries'])} "
f"individual shard-split operations, each moving ~1/{large['num_primaries']} of the data "
f"-- {small['num_primaries'] / large['num_primaries']:.0f}x FEWER migrations needed, but each one relocates "
f"{small['num_primaries'] / large['num_primaries']:.0f}x more data and briefly holds "
f"{100 / large['num_primaries']:.1f}% of total capacity at reduced redundancy mid-split")
Output:
target: 400,000 QPS sustained, 1 replica/primary
-- many small instances (20 primaries, 40 instances total) --
compute: $19,856/mo
ops overhead: $6,000/mo
TOTAL: $25,856/mo
blast radius if one primary fails: 5.0% of write capacity
-- fewer large instances (4 primaries, 8 instances total) --
compute: $19,856/mo
ops overhead: $1,200/mo
TOTAL: $21,056/mo
blast radius if one primary fails: 25.0% of write capacity
fewer-large saves $4,800/mo (18.6%) at this target, driven entirely by the 32 fewer instances' worth of fixed ops overhead -- compute cost itself is ~equal by construction (linear pricing assumption).
-- resharding friction: operations needed to DOUBLE total capacity --
many-small (20 primaries): 20 individual shard-split operations, each moving ~1/20 of the data
fewer-large (4 primaries): 4 individual shard-split operations, each moving ~1/4 of the data -- 5x FEWER migrations needed, but each one relocates 5x more data and briefly holds 25.0% of total capacity at reduced redundancy mid-split
At this specific target (400,000 QPS, "small" instances rated at 20,000 QPS each versus "large" instances rated at 100,000 QPS each, 5x the capacity at 5x the hourly price, one HA replica per primary), fewer-large saves about 18.6 percent per month, entirely from having 32 fewer instances' worth of fixed operational overhead to carry, since compute cost is identical by construction (both plans total $19,856/mo in compute). But fewer-large also concentrates 25 percent of total write capacity behind a single instance failure, versus 5 percent for many-small, a 5x difference in blast radius for an 18.6 percent cost saving.
Failover impact
Blast radius scales as 1 divided by the number of primaries: many-small spreads a single-instance failure across a larger denominator, so losing one instance costs less of the total in relative terms, at the (usually acceptable) cost of more moving parts to monitor. This is the sharpest trade-off in the whole model and the one most likely to actually decide the answer: a 25 percent capacity loss during a failover (the process of detecting a dead instance and promoting a replacement to take over its work) is a very different incident than a 5 percent one, and that difference does not show up on the cost line at all.
Migration complexity
Doubling total capacity by splitting each primary in two costs one shard-split operation per existing primary: many-small needs more individual operations (more shards to split), but each one moves a smaller fraction of the data and holds a smaller fraction of total capacity at reduced redundancy during the split. Fewer-large needs fewer operations, but each one is a bigger, riskier undertaking: more data in flight per split, and a larger fraction of total capacity briefly exposed while that split is in progress.
Applying the model to a decision
- Choose fewer-large when: operator time is the actual scarce resource (a small team managing the cluster), the workload can tolerate a real capacity dip during a single-instance failure (redundancy and fast failover elsewhere in the stack absorb it), and the per-instance overhead assumption in the model reflects real toil, not a number picked to make the comparison come out a certain way.
- Choose many-small when: the workload has a tight availability commitment where a 20+ percent capacity loss during a routine failure is unacceptable, resharding needs to happen incrementally and safely (smaller, lower-risk operations), or the engine is licensed per core with a meaningful PER-INSTANCE base fee on top, in which case more instances directly costs more in licensing, not just more operational overhead, and the model above would need that term added explicitly before trusting its conclusion.
- The model's answer is only as good as its inputs. Before applying it, verify the operational-overhead-per-instance figure against real toil (ticket volume, on-call load) rather than a guess, and verify whether the specific engine's licensing model is truly per-core (favors neither topology directly) or has a per-instance component (favors fewer-large).
Trade-offs and pitfalls
- Do not let "compute cost is roughly equal" get mistaken for "the two options are roughly equivalent." The blast-radius and migration-granularity differences are real and often matter more than the modest overhead saving.
- The linear-pricing assumption in the model is a simplification stated explicitly, not a verified fact for every vendor and instance family; re-derive the compute-cost lines with the actual current pricing for the instance types under consideration before using this for a real budget decision, since cloud list pricing changes and sometimes is not perfectly linear at the largest instance sizes.
- A cost model that never gets revisited goes stale. Redundancy requirements, instance pricing, and the team's actual operational capacity all drift over time; treat this as a model to re-run periodically, not a one-time decision.
Describe three common sharding strategies: range-based sharding, hash-based sharding, and directory (lookup)-based sharding. For each strategy explain how keys are mapped to shards, typical advantages, failure modes (hotspots, rebalancing cost), and a concrete scenario where it is usually preferred.
Sample Answer
Direct answer
The three standard sharding strategies map keys to shards by range, by hash, or by an explicit lookup table. Range-based sharding assigns contiguous key ranges to shards (great for range queries, bad for monotonically increasing keys, which all land on the newest shard). Hash-based sharding runs the key through a hash function and assigns by the result (excellent even distribution, but destroys range-query locality and makes resharding expensive unless you use consistent hashing). Directory-based (lookup) sharding keeps an explicit key-to-shard mapping table (maximum flexibility to rebalance arbitrarily, at the cost of that lookup table becoming a dependency and potential bottleneck itself).
Structured elaboration
| Strategy | How keys map to shards | Advantages | Failure modes | Prefer it when |
|---|---|---|---|---|
| Range-based | Each shard owns a contiguous range of key values (A-F, G-M, ...) | Range scans (customer_id BETWEEN X AND Y) stay on one or a few shards; simple to reason about | Hotspots on monotonically increasing keys (all new writes hit the newest/highest range); ranges grow unevenly, forcing splits | Query patterns are range-scan-heavy (analytics over a sorted key, time-ordered access) and the key is not itself monotonic in a way that concentrates writes |
| Hash-based | hash(key) mod N (or a hash ring) picks the shard | Near-uniform distribution regardless of the key's natural skew; simple to implement | Range queries now have to fan out to every shard, since hashing destroys ordering; naive mod N remaps most keys when N changes (consistent hashing, covered below, fixes this specific failure mode) | Access is point-lookup-dominated (WHERE user_id = ?) rather than range-scan-dominated, and even write distribution matters more than range locality |
| Directory-based (lookup) | An explicit table or service maps each key (or key range) to a shard | Maximum flexibility: can rebalance arbitrarily, move individual hot keys to their own shard, without any mathematical constraint on the mapping | The directory itself is a new critical-path dependency: it needs its own availability and consistency story, and every request pays a lookup (usually cached) before it can route | You need fine-grained, ad hoc control over placement, e.g. moving one specific hot tenant to dedicated hardware, that a formula-based scheme (range or hash) cannot express |
Consistent hashing, since it changes the "hash-based" trade-off: plain hash(key) mod N means adding or removing a shard changes almost every key's target shard, forcing a near-total data reshuffle. Consistent hashing places both shards and keys on a ring (a circular hash space) and assigns each key to the next shard clockwise from its position; adding or removing one shard only remaps roughly 1/N of the keys, the ones between the changed shard and its neighbor, not the whole dataset. This is why hash-based sharding is viable in practice for systems that expect to add and remove shards over time.
A concrete strategy for a wide-column store (Cassandra), a NoSQL database that stores each partition as a set of columns grouped under a row key instead of a relational table's fixed rows: Cassandra's partition key (the sharding key) determines placement via consistent hashing with virtual nodes (each physical node owns many small ranges on the ring, not one big contiguous range, which keeps rebalancing smooth when nodes join or leave). Within one partition, a separate clustering column controls the on-disk sort order of rows, which is what makes range queries within a partition fast even though the partition key itself is hash-distributed. The standard technique for time-series data on Cassandra is time bucketing: instead of a partition key of just sensor_id (which would grow one partition unboundedly forever), use a composite partition key like (sensor_id, date_bucket) so each partition covers one sensor's readings for one day (or hour), with timestamp as the clustering column ordering rows within that bucket. This keeps individual partitions bounded in size while still supporting "give me sensor X's readings for date Y, in order" as a single-partition, clustering-column-ordered read.
Implementing and resharding in a managed cloud environment with minimal downtime. Managed sharded services (MongoDB Atlas, Vitess, Citus, or a cloud-managed Cassandra/DynamoDB) typically expose resharding as an online, background operation rather than a maintenance-window event: they migrate chunks/ranges from the over-loaded shard to new or under-loaded ones incrementally, keep both the old and new location briefly dual-writable or use change-data-capture to catch up the new location, then atomically cut over routing for that range. The operator's job shifts from "execute a migration script" to "choose a shard key that keeps the automatic balancer's job easy" (even distribution, no reliance on the current shard count) and to monitor that the background rebalance is proceeding rather than falling behind write volume.
An SRE-facing shard-key evaluation checklist (useful framing when you're the one who gets paged, not just the one who designs the schema): does the key distribute load evenly under the real (not assumed) traffic shape; does a single key's data fit comfortably within one shard's storage and throughput budget; can the key be rebalanced without an application-visible migration window; and does the routing mechanism (hash function, range table, or directory) have its own monitoring and failure mode understood, separately from the data plane itself.
Worked example
import hashlib, statistics
N_SHARDS, N_KEYS = 8, 100_000
def hash_shard(user_id, n_shards):
h = hashlib.md5(str(user_id).encode()).digest()
return int.from_bytes(h[:8], "big") % n_shards
counts = [0] * N_SHARDS
for uid in range(1, N_KEYS + 1):
counts[hash_shard(uid, N_SHARDS)] += 1
mean = statistics.mean(counts); stdev = statistics.pstdev(counts)
print(f"hash-based: counts={counts}")
print(f"mean={mean:.1f} stdev={stdev:.1f} ({stdev/mean:.1%} of mean)")
# Range-based on the same key space, but only the LAST 5000 ids minted this week
range_width = N_KEYS // N_SHARDS
range_counts = [0] * N_SHARDS
for uid in range(N_KEYS - 5000, N_KEYS + 1):
shard = min((uid - 1) // range_width, N_SHARDS - 1)
range_counts[shard] += 1
print(f"range-based, this week's writes only: {range_counts}")
print(f"top shard's share of this week's writes: {max(range_counts)/sum(range_counts):.1%}")
hash-based: counts=[12227, 12574, 12683, 12547, 12541, 12531, 12396, 12501]
mean=12500.0 stdev=127.1 (1.0% of mean)
range-based, this week's writes only: [0, 0, 0, 0, 0, 0, 0, 5001]
top shard's share of this week's writes: 100.0%
Same 100,000-key space, same 8 shards: hash-based sharding spreads the full dataset within 1.0% of perfectly even. Range-based sharding, applied to a monotonically increasing id, sends 100% of this week's new writes to a single shard (the top of the range), because every new id is by definition higher than every existing one. This is the concrete mechanism behind "range-based sharding hotspots on monotonic keys."
Trade-offs & pitfalls
- Picking hash-based sharding and then being surprised range queries got slow. A hash-sharded key has no relationship between neighboring key values and neighboring shards, so
WHERE customer_id BETWEEN X AND Yhas to fan out to every shard and merge, exactly the scatter-gather cost range-based sharding was designed to avoid. - Picking range-based sharding on a naturally monotonic key without adding jitter. The worked example above is not a corner case, it's what happens by default to any auto-incrementing id or a
created_at-derived key under range sharding; the standard fix is either switching to hash-based for that key, or reversing/bucketing the monotonic component (e.g. bit-reversal, or a coarse date bucket combined with a hash suffix) before using it as the range boundary. - Directory-based sharding's flexibility is not free. Every request now depends on a lookup (even if cached) succeeding and being consistent; a stale or incorrect directory entry silently routes a request to the wrong shard, which is a different and often harder-to-detect failure mode than a hash function simply producing the "wrong" (but deterministic and reproducible) answer.
Describe the difference between sharding and replication. Give specific examples of how each affects availability, latency, capacity, and consistency in a production database architecture. When would you choose one over the other for a new service?
Sample Answer
Direct answer
Replication and sharding solve two different bottlenecks. Replication copies the same data onto multiple nodes, so you can lose a node without losing data and can spread read traffic across copies. Sharding (also called partitioning) splits a dataset by a key so that no single node holds all of it, which is what lets total write throughput (the rate of writes a system can accept per second) and total storage grow past what one machine can do. Most systems that actually reach scale do both at once: each shard is itself a replicated set, so replication buys durability and read capacity while sharding buys write and storage capacity.
How each one moves the four dials
| Dial | Replication (leader + followers) | Sharding (by key) |
|---|---|---|
| Availability | Improves it: a follower can be promoted if the leader dies | Neutral to slightly worse on its own: losing a shard still takes its slice of data offline unless each shard is also replicated |
| Latency | Reads can go to a nearby follower, lowering read latency; writes still funnel through one leader | Writes to different keys go to different nodes in parallel, so write latency does not grow with total load, but a query touching multiple shards now pays fan-out latency |
| Capacity | Adds read capacity (more copies to read from) and durability; does not add write or storage capacity, since every follower stores the full dataset | Adds write capacity and storage capacity, since each shard only holds and only accepts writes for its own slice |
| Consistency | Followers can lag behind the leader (replication lag), so a reader on a follower can see stale data | Queries that only touch one shard keep whatever consistency that shard offers; queries spanning shards (a cross-shard join or a multi-shard transaction) lose easy strong consistency unless you add a distributed transaction protocol |
Worked example
A new internal task-tracking service launches at 500 writes/second and 50 GB of data. A single well-provisioned primary can hold 50 GB on local NVMe (a fast solid-state storage interface) storage without strain and has enormous headroom below any write-throughput ceiling at 500 writes/second, so one primary plus two read replicas (for durability and to take read load off the primary) is the right architecture. Sharding here would add a routing layer, give up easy transactions across tasks, and buy zero capacity headroom the team actually needs yet.
Contrast that with a global ad-serving event ledger ingesting 2 million writes/second across 500 TB. No single primary, however well tuned, accepts 2 million writes/second or stores 500 TB on one node, so the dataset must be split across many shards by a key such as account_id. Each of those shards is then, separately, replicated (a leader and followers per shard) so a single shard's leader dying does not take that slice of the ledger offline.
Trade-offs and pitfalls
Reach for replication first: it is the simpler system, requires no routing layer, and preserves single-node transactions and joins. Only shard once you can show the actual bottleneck is a single primary's write throughput or a single node's storage ceiling, not "the data might get big someday." The common wrong turn is sharding pre-emptively: it adds a routing tier, a rebalancing story as shards grow unevenly (hot-key skew), and the loss of cheap cross-entity joins and transactions, none of which pays for itself until the write or storage ceiling is a real, measured problem. The other common wrong turn is the reverse: piling more and more read replicas onto a service that is actually write-bound, which never touches the real bottleneck because replication does not add write capacity.
Unlock Full Question Bank
Get access to all Replication, Partitioning, and Sharding interview questions and detailed answers.
Sign in to ContinueJoin thousands of developers preparing for their dream job.