Data Partitioning and Sharding Questions

Techniques and operational practices for horizontally partitioning data across multiple database instances or storage nodes to achieve scale, improve performance, and manage growth. Includes selection and design of partition and shard keys to evenly distribute load and avoid hotspots, with range based, hash based, and directory based approaches and consistent hashing mechanisms. Covers handling uneven distribution and data skew, hotspot detection and mitigation, and the impact of partitioning on query patterns such as joins and cross shard queries. Explains implications for transactions and consistency, including transactional boundaries that span partitions and approaches to distributed transactions and compensation. Describes resharding and online data migration strategies, rolling rebalances, and methods to minimize downtime and data movement. Emphasizes operational concerns including shard management, automation, monitoring and alerting, failure recovery, and performance tuning. Discusses trade offs between simplicity, latency, throughput, and operational complexity and highlights considerations for both transactional and analytical workloads, including routing, caching, and coordination patterns.

MediumTechnical

75 practiced

Given expected growth to 100M users and average user data size of 5MB, design a capacity plan that estimates number of shards, storage needs, and per-shard throughput if each user generates 1 write per day and reads average 50 requests/day. Show calculations and assumptions clearly.

Sample Answer

Assumptions (explicit):- Users target: 100M- Avg user data stored: 5 MB → total primary user data = 100M * 5 MB = 500,000,000 MB = 500 TB- Write pattern: 1 write/user/day- Read pattern: 50 reads/user/day- Peak factor: 2x (to account for hourly peaks/spiky traffic)- Availability overhead + replicas: 3x replication (master + 2 replicas) for durability- Per-shard practical storage capacity: 2 TB usable (after filesystem overhead, indexes); per-shard max steady throughput: 500 writes/sec and 10,000 reads/sec (example DB like horizontally sharded document store; adjust per tech)

Step 1 — Storage needed (raw and replicated):- Raw logical data = 500 TB- With 3x replication → physical storage = 500 TB * 3 = 1,500 TB = 1.5 PB- Add 20% headroom for indexes/metadata/retention → 1.5 PB * 1.2 = 1.8 PB

Step 2 — Number of shards for capacity:- Per-shard usable storage = 2 TB → shards_for_storage = 1,800 TB / 2 TB = 900 shardsRound up → 900 shards

Step 3 — Throughput calculations (QPS):- Daily writes total = 100M writes/day = 100,000,000 / 86,400 ≈ 1,157 writes/sec average- Daily reads total = 100M * 50 = 5,000,000,000 reads/day → 5e9 / 86,400 ≈ 57,870 reads/sec average- Apply peak factor 2x → peak writes ≈ 2,314 WPS, peak reads ≈ 115,740 RPS

Per-shard throughput:- With 900 shards: - Peak writes per shard = 2,314 / 900 ≈ 2.6 WPS - Peak reads per shard = 115,740 / 900 ≈ 128.6 RPS

Compare to per-shard capacity (assumed 500 WPS, 10k RPS) — well within limits, giving headroom for growth and hotspots.

Notes and trade-offs:- If per-shard storage or throughput differs (e.g., larger nodes), shard count changes proportionally.- Consider range vs. hash sharding to avoid hot keys; add caching (CDN / Redis) to lower read QPS from DB.- Consider retention policies: if historical data accumulates, storage needs grow — use colder tiers (S3/segment stores) and keep hot 3–6 months in DB to reduce shard count.- Monitor and allow autoscaling; plan for at least 2x extra shards for rebalancing/maintenance.

HardTechnical

135 practiced

Implement (pseudocode allowed) an algorithm that assigns virtual nodes to physical shards to balance expected load when physical shards have different capacities (heterogeneous hardware). Explain how you compute the number of vnodes per shard and how you verify balanced distribution.

Sample Answer

Approach: compute each physical shard’s share of total capacity and assign vnodes proportionally, then place vnodes on the consistent-hashing ring with randomized, deterministic offsets; apply integer rounding with a residual-distribution step (largest remainder) to ensure total vnode count matches target. Verify by simulating expected load per shard using key-to-vnode mapping and checking deviation against a tolerance.

Pseudocode (Python-like):

python

def assign_vnodes(shards, total_vnodes):
    # shards: list of (shard_id, capacity)
    # total_vnodes: desired total virtual nodes across cluster
    total_capacity = sum(c for (_, c) in shards)
    # raw fractional allocation
    raw = [(s, total_vnodes * c / total_capacity) for (s,c) in shards]
    # floor allocation and residuals for largest-remainder
    alloc = {s: int(fr) for (s,fr) in raw}
    remainder = [(s, fr - alloc[s]) for (s,fr) in raw]
    remaining = total_vnodes - sum(alloc.values())
    # give extra vnodes to highest remainders
    for s,_ in sorted(remainder, key=lambda x: -x[1])[:remaining]:
        alloc[s] += 1
    # generate vnode positions deterministically per shard
    vnode_positions = []
    for s,count in alloc.items():
        # use hash(shard_id + idx) to place on ring [0,1)
        for i in range(count):
            pos = hash_to_unit_interval(f"{s}-{i}")
            vnode_positions.append((pos, s))
    vnode_positions.sort()
    return vnode_positions, alloc

Verification:- Simulate N random keys, map each key -> ring position -> nearest vnode -> shard, count keys per shard.- Expected share = total_keys * (capacity_shard / total_capacity). Compute relative error = |observed - expected|/expected.- Accept if max relative error <= tolerance (e.g., 1-2% plus quantization effects).

Complexity:- Allocation: O(M log M) for sorting residuals (M shards).- Placement: O(total_vnodes log total_vnodes) to sort positions.- Verification simulation: O(N log V) (binary search on vnode positions) or O(N) with interval indexing.

Edge cases & improvements:- Very small shards may get zero vnodes; enforce min_vnodes or shard exclusion policy.- If total_vnodes too small, quantization causes large error — increase total_vnodes for granularity.- For dynamic capacity changes, reassign only necessary vnodes (move minimal set) using consistent hashing techniques to reduce data movement.

MediumTechnical

87 practiced

Describe a programmatic approach to detect data skew across shards using production telemetry. Explain what signals you'd compute, how you'd normalize for traffic patterns, and one automated mitigation you could apply when a shard is overloaded.

Sample Answer

Approach: continuously compute per-shard telemetry, normalize against expected patterns, detect outliers, and trigger an automated mitigation (traffic reroute or shard split) when a shard is overloaded.

Signals to compute:- QPS (requests/sec) per shard- CPU% and memory% per shard- P95/P99 latency and error rate- Queue/backlog depth (pending tasks)- Disk I/O or hot-key counters (if applicable)Compute both raw and rate-of-change metrics.

Normalization and detection:- Maintain rolling baseline per-shard for each metric using an exponential moving average (EMA) or 24-hour hourly profile to capture diurnal patterns.- Convert current metric to a normalized anomaly score (z-score or robust median absolute deviation) vs. baseline: score = (value - baseline_mean) / baseline_std- Weight signals (e.g., latency & QPS higher) and compute composite anomaly score.- Use windowed aggregation and require sustained breach (e.g., score>3 for 5 minutes) to avoid flapping.

Example detection loop (simplified):

python

import time, statistics

def detect_anomaly(shard_metrics, baseline_stats, weights, threshold=3):
    scores = {}
    for m, val in shard_metrics.items():
        mean, std = baseline_stats[m]
        std = max(std, 1e-6)
        scores[m] = (val - mean) / std
    composite = sum(weights[m]*scores[m] for m in scores) / sum(weights.values())
    return composite > threshold, composite

Automated mitigation (one option — proportional traffic reroute):- When a shard is overloaded, adjust gateway/router weights to route a fraction of new requests to other healthy shards or read replicas.- Steps: mark shard “degraded”, compute reroute fraction proportional to overload score, apply gradual weight shift (e.g., 10% every 30s) with circuit-breaker and rollback if downstream capacity saturates.- Safety: require capacity checks on target shards, cooldown period, alerting and human-in-the-loop fallback.

Why this works:- Normalization removes expected traffic cycles, composite scoring reduces false positives, and gradual reroute mitigates outage risk while giving time for more heavy-weight fixes (reshard/migrate).

HardSystem Design

85 practiced

Describe online and offline strategies to change the shard key for an existing table containing hundreds of millions of rows. Provide a high-level migration plan that minimizes downtime and preserves data correctness, and explain how you would test and roll back the change.

Sample Answer

Requirements & constraints:- Change shard key for a production table with hundreds of millions of rows.- Minimize downtime (near-zero), preserve correctness, support reads/writes during migration, allow rollback, and ensure low performance impact.

High-level approaches:1. Offline (bulk rewrite): export → transform → import. Simple and fast overall but requires maintenance window; higher downtime and heavy I/O; use only if downtime acceptable.2. Online (dual-write + backfill): preferred. Serve reads/writes from old shard layout while backfilling data into new shard layout in background; switch reads atomically when ready.

Recommended online migration plan (phased):1. Prep - Add new sharding metadata, routing layer capable of conditional routing (by tenant, time, or migration state). - Create new shard topology and allocate capacity.2. Dual-write - Update application/service to write to both old and new shard locations for all writes (idempotent writes). - Make dual-write behind a feature flag for safe rollout.3. Backfill (consistent copy) - Launch backfill workers that scan old shards and write into new shards. - Use change-data-capture (CDC) or oplog to capture concurrent updates; apply them to new shards in order to preserve causality. - Track per-row version or log positions and maintain progress checkpoints.4. Verify - Run row-level and aggregate consistency checks (counts, checksums, sample hashes) for each shard range. - Run application read-tests against new shards using canary traffic.5. Cutover - Flip routing: route reads to new shards (can be gradual by customer or shard-range). - Keep dual-write for a short verification window.6. Finalize - Stop dual-write, decommission old shards when safe, reclaim storage.

Data correctness techniques:- Use per-row monotonic version (last-modified timestamp or change-sequence) to resolve conflicts.- Make writes idempotent and include write-nonce to avoid duplicates.- Use transactional writes where available (two-phase commit if cross-shard atomicity needed) or eventual consistency with compensating reconciliation.

Testing strategy:- Unit & integration tests for dual-write, idempotency, and routing.- Stage run on a copy of production data at scale or replay production traffic in a sandbox.- Chaos tests: kill workers, pause CDC, simulate network partitions.- Consistency tests: full-table row counts, partition-level checksums, sampled row-by-row compare, and application-level end-to-end scenarios.- Performance testing to ensure acceptable latency and resource usage.

Rollback plan:- If inconsistencies or regressions detected before cutover: stop backfill, revert routing and feature flags; continue using old shards.- If issues after partial cutover: route affected traffic back to old shards, keep dual-write active while fixing backfill or bug, re-run backfill for the gap (use CDC offsets).- Automate rollback scripts to restore routing and disable writes to new shards; ensure idempotency lets re-run safely.

Operational considerations & trade-offs:- Dual-write increases write latency and resource usage; mitigate by batching and async workers.- CDC + backfill is complex but minimizes downtime.- Offline bulk rewrite simpler but requires large window and careful capacity planning.- Monitor: per-shard lag, error rates, consistency metrics, and storage usage.

This plan balances near-zero downtime, preserves correctness via CDC+versioning, provides test/rollback paths, and limits blast radius via incremental cutover.

MediumBehavioral

121 practiced

Tell me about a time when you identified and resolved a production hotspot or data skew issue. Describe the context, your analysis, the actions you took (technical and operational), and the outcome. Use the STAR method (Situation, Task, Action, Result).

Unlock Full Question Bank

Get access to hundreds of Data Partitioning and Sharding interview questions and detailed answers.

Join thousands of developers preparing for their dream job.