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

98 practiced

Given a mapping of ranges to shards and a target capacity distribution, describe a greedy algorithm to compute range splits and target assignments that minimize total data movement when rebalancing. Explain the algorithm steps, complexity, limitations, and how you'd improve results in practice.

Sample Answer

Approach (high level)Build splits so each final range is assigned wholly to one shard and the sum of capacities per shard matches targets while minimizing bytes moved. Greedy idea: walk the keyspace left-to-right, at each step emit the largest contiguous subrange that can be assigned to the current shard without exceeding its target capacity (subject to rounding), preferring to reuse existing shard owners to avoid movement.

Algorithm steps1. Precompute for each contiguous input range its size and current owner shard.2. Compute target byte quotas per shard from total bytes × target distribution.3. Initialize pointer at start of keyspace and remaining_quota[shard] = target_quota.4. While pointer not at end: a. For the current position consider contiguous input ranges forward. For each candidate end, compute candidate_size and a movement_cost = candidate_size × (1 if current owner != candidate_assigned_shard else 0). b. Select shard S that (i) has remaining_quota >= candidate_size (or the largest that fits), and (ii) minimizes movement_cost; break ties by maximizing reuse or filling quota. c. Emit split [pos, end) assigned to chosen S; subtract candidate_size from remaining_quota[S]; advance pointer to end.5. If quotas remain after a pass, allow small overshoot/fractional last ranges or run a second pass to adjust.

Complexity- Let R = number of input ranges. Inner step may scan forward; naive worst-case O(R^2). With prefix sums and greedy lookahead limited to next K ranges, complexity ~O(R·K). Using efficient pointers and only considering boundaries yields O(R).

Limitations- Greedy is local: can produce suboptimal global assignment (e.g., small ranges cause fragmentation and force large moves later).- Sensitive to initial ordering and rounding of quotas.- Assumes sizes and targets static; doesn't account for costs of moving many small fragments vs one big chunk.

Practical improvements- Pre-merge tiny adjacent ranges to reduce fragmentation.- Run local refinement: allow swapping neighboring assignments if it reduces movement while maintaining quotas.- Use min-cost flow formulation for optimal result (nodes = range segments, capacity = size, cost = movement) and solve for moderate R.- Heuristics: allow lookahead window, prioritize reusing current owners, penalize splitting ranges, and post-process with simulated annealing or ILP for critical rebalances.- Validate with staging run, enforce max per-shard transfer limits and schedule low-load windows for heavy moves.

EasyTechnical

104 practiced

You have an events table for analytics: events(event_id, user_id, event_type, event_ts TIMESTAMP, properties JSON). Typical queries: time-window aggregations by day, funnels by user, occasional user-level joins. Propose a partitioning and clustering strategy for a cloud data warehouse (e.g., BigQuery or Snowflake) and justify your choices.

Sample Answer

Situation: You need a practical partitioning/clustering layout optimized for the listed access patterns: daily time-window aggregations, user-level funnels, and occasional user joins.

Recommendation (BigQuery / Snowflake-agnostic guidance, with notes):

1) Partitioning- Partition by event_ts DATE (day) — i.e., DATE(event_ts) partitioning (BigQuery: ingestion-time optional but prefer event_ts; Snowflake: use RANGE on DATE_TRUNC('day', event_ts) or rely on micro-partitions and a date column filter).- Set a sensible partition retention/expiration (e.g., 365–1095 days) if old data is rarely queried.Justification: Most queries filter on time windows; daily partitioning gives strong partition pruning and dramatically reduces bytes scanned/cost.

2) Clustering / Clustering Key / Sort Key- BigQuery: cluster the partitioned table on (user_id, event_type). Optionally add event_ts or a hashed_user_bucket as a 3rd key if you need better distribution.- Snowflake: define a cluster key on (user_id, event_type) only if automatic micro-partitioning isn’t pruning well; otherwise rely on natural micro-partitioning and use clustering for very large tables with skew.Justification: Funnels and user-level joins filter by user_id and event_type; clustering colocates rows for same user/event_type within partitions so queries scan fewer blocks and improve performance.

3) Additional practical items- Limit cluster key cardinality: user_id is high-cardinality but funnel queries are per-user, so benefit outweighs cost; mitigate hotspotting by using a hashed_user_bucket (e.g., user_id % 256) plus user_id if necessary.- Keep clustering columns stable and few (BigQuery recommends ≤4 columns).- Regularly run table re-clustering / optimize/compact operations (Snowflake: RECLUSTER/ALTER; BigQuery: clustering is automatic but may need periodic copy to re-cluster).- Consider materialized views or pre-aggregations for heavy daily metrics; store heavy JSON properties parsed into columns for frequent predicates.- Monitor query patterns and adjust: if joins are heavy on user_id and device_id, add device_id to clustering.

Trade-offs and edge cases- Very high-cardinality clustering (only user_id) can reduce pruning benefits; use hashed buckets if skewed active users cause hotspots.- Coarser partitioning (monthly) reduces partition count but increases scanned data — stick with daily for typical analytics.

Result: Day partitions minimize scan for time filters; clustering on user_id + event_type speeds funnels and joins. This balances cost, query latency, and maintenance for an analytics events table.

HardSystem Design

76 practiced

Design a global multi-region sharding and replication strategy for user-owned financial data requiring low-latency regional reads and strong consistency for transactions. Constraints: 100M users, 1B daily operations, data residency rules in some regions. Explain shard placement, leader selection, replication model (sync vs async), routing, and failover procedures while balancing latency and consistency.

Sample Answer

Requirements & constraints:- Functional: user-owned financial records; low-latency regional reads; strong consistency for transactions (ACID per user account).- Scale: 100M users, 1B ops/day.- Non-functional: regional data residency for some jurisdictions, <= single-digit ms read latency regionally, durability, predictable failover.

High-level approach:- Use user-centric partitioning (shard by user_id) so each user’s financial state lives in a single shard to ensure local serializability.- Deploy regional shard groups: for each region where users live, maintain shards located primarily in that region (data residency). For globally-mobile users, place shard in their "home" region (or region of account creation) with optional secondary replicas elsewhere per policy.

Shard placement:- Hash(user_id) → shard_id; maintain a consistent-hash ring with range mappings to physical shard instances per region.- Provision shards sized for throughput (e.g., 50–200k users / shard) and scale via re-sharding using virtual nodes to reduce movement.

Leader selection & replication model:- Each shard is a Raft/Percolator-like group: one leader (primary) and N-1 replicas.- For regions with residency requirements: leader and majority of replicas located inside the region.- Replication: synchronous (strong) replication to a majority quorum for transaction commits (leader writes to followers and waits for majority ack before commit) — ensures linearizability for writes.- For read latency: enable follower reads only when bounded-staleness is acceptable; otherwise route reads to leader. To get low-latency reads regionally, place a read-only locallearner replica that receives async replication but only serves reads for non-critical queries; critical balance: financial transactional reads go to leader or quorum reads with timestamp-based leader leases.

Routing:- API gateway or routing layer maps user_id → shard metadata in a global config service (consistent, cached). Clients send writes to shard leader; reads go to leader for strong consistent reads. For local low-latency dashboards, prefer leader if leader is local; if leader remote but quorum-local exists, use read-after-quorum pattern or perform single-hop forwarding to leader.- Use a placement/metadata service (distributed config in Paxos/Raft) for live leader location and shard map.

Failover procedures:- Leader failure: Raft automatic leader election within shard group — elect local replica that has up-to-date log entries and respects residency constraints (tie-breaker prefers local/regional replica).- Cross-region outage: if entire region down and its shards must remain available, promote replicas outside region only if compliance permits (flagged per-shard). Otherwise, mark affected shards read-only and route transactions to legal alternate region after manual/automated policy checks.- Rolling upgrades and maintenance: use leader transfer to preserve locality before draining nodes.

Balancing latency & consistency:- Use quorum-sync writes (leader + majority) to guarantee consistency; prevent stale reads by routing financial reads to leader or using linearizable read path (lease/read-index).- For analytics pipelines, use async change-data-capture (CDC) from the leader's WAL to feed eventual-consistent data lakes; ensure transactional boundaries by streaming commit markers.

Operational considerations:- Monitoring: per-shard QPS, replication lag, leader location, election rate.- Autoscaling: add shards via virtual nodes, rebalance gradually.- Compliance: per-region key management (KMS), log and backup location controls, per-shard residency metadata controlling replica placement and failover policies.

Trade-offs:- Strong consistency (sync majority) increases write latency vs async; mitigated by co-locating leader near majority of active users and using sharding to keep per-shard latency low.- Cross-region failover vs compliance: automated failover may violate residency — enforce policy-driven manual escalation for restricted shards.

This design provides per-user strong consistency for transactions, low-latency regional reads by co-locating leaders or using safe read strategies, and compliance-aware replica placement and failover.

EasyTechnical

78 practiced

Explain consistent hashing conceptually and why it reduces data movement during resharding compared to a simple modulo-based shard mapping. Describe the effect of adding or removing a node on key-to-node mappings and why virtual nodes might be used.

HardSystem Design

147 practiced

Design a system to provide ACID-like guarantees across multiple shards for an order fulfillment flow without 2PC. Use sagas with compensating transactions, durable saga logs, idempotent operation design, and strong primitives where necessary. Explain how you ensure at-least-once or exactly-once semantics, how to model compensations, and how to surface unresolved sagas to operators.

Sample Answer

Requirements & constraints:- Cross-shard order fulfillment (inventory shard, payments shard, shipping shard) with ACID-like guarantees but no 2PC.- Use sagas with durable logs, compensating transactions, idempotency, and strong primitives when needed.- Provide observable unresolved sagas and operator tooling.

High-level architecture:- Saga orchestrator service (stateless, scalable) + durable Saga Log (append-only, highly-available store like DynamoDB, Bigtable, Kafka compacted topic).- Per-service transaction endpoints that accept an operation idempotently and expose compensating endpoints.- Reliable message bus (Kafka or Cloud Pub/Sub) for at-least-once delivery and retries.- Monitoring/alerting dashboard and operator UI to surface unresolved sagas.

Saga flow:1. Create saga record in Saga Log with unique saga_id, current step, payload, deadline, retry policy.2. Orchestrator issues step 1 (reserve inventory) by writing a command to the service topic with saga_id and op_id.3. Target service performs idempotent action (check op_id), writes local state, a local event, and replies success/failure. It persists compensating info (e.g., reservation token).4. Orchestrator updates Saga Log atomically (compare-and-set) to next step and issues next command (charge payment).5. On any step failure, orchestrator executes compensating steps in reverse order by issuing compensating commands (cancel reservation, refund).

Ensuring at-least-once vs exactly-once:- At-least-once: achieved by retrying commands until the orchestrator sees success in Saga Log. Services must implement idempotency using op_id and idempotency table (bounded TTL) and make operations idempotent e.g., PUT reservation with idempotency-key.- Exactly-once effect (from business POV): combine idempotency + strong primitives: - Use conditional writes for local updates (DB unique constraint on op_id) so duplicate commands are no-ops. - For payments, prefer provider APIs supporting idempotency tokens or ledger append-only entries. - For inventory, use compare-and-set (decrement if available) with op_id to avoid double-debit. - Where necessary, use single-writer per shard or leader election to serialize access.

Modeling compensations:- Compensations are explicit operations stored with the forward step in the Saga Log: {step_id, forward_cmd, compensation_cmd, retry_policy, deadline}.- Compensations should be compensating actions that restore invariants (e.g., release reservation, issue refund). They must be idempotent and persistent (store compensation outcome).- For non-reversible actions (shipping label purchased), compensation may be a business-level compensation (refund + credit note) and must be flagged as manual-required if automated compensation impossible.

Durable saga log & consistency:- Use a durable, strongly-consistent store for Saga Log supporting CAS/optimistic locking so orchestrator updates are atomic and crash-recoverable.- Orchestrator reads/writes Saga Log to drive state machine; on restart it resumes in-progress sagas.- For high throughput, partition saga log by saga_id hash.

Failure handling & retries:- Exponential backoff with bounded retries for transient failures.- If compensation or forward step repeatedly fails beyond threshold, mark saga as "requires manual intervention" and escalate.

Surfacing unresolved sagas to operators:- Saga lifecycle states: PENDING, IN_PROGRESS, COMPLETED, COMPENSATING, FAILED_AUTOMATED, FAILED_MANUAL.- Instrument metrics (counts, latencies), logs, and traces (propagate saga_id).- Operator UI/alerts: list sagas in FAILED_AUTOMATED or FAILED_MANUAL with details: steps completed, last error, payload, timestamps, suggested actions (run compensation, retry step, or manual refund).- Provide automated runbook actions in UI to trigger compensations, replay specific steps, or mark resolved. All operator actions append to Saga Log for audit.

Trade-offs:- No 2PC reduces global blocking and cross-shard locks but requires careful compensation modeling and potential eventual consistency.- Strong primitives (conditional writes, idempotency tokens) add complexity but are necessary for exactly-once business effects.- Manual intervention required in edge cases where compensation cannot be automated.

Observability & governance:- Enforce schema for forward/compensate commands, require per-service SLA for compensations.- Periodic reconciliation jobs to detect state drift (compare saga outcomes with shard state) and auto-enqueue remediation sagas.

This design gives ACID-like guarantees at the business level: atomicity via forward+compensate, durability via Saga Log, isolation via idempotent, conditional local writes, and consistency via orchestrated sequencing and reconciliations—without blocking distributed 2PC.

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.