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.

EasyTechnical

79 practiced

When should a team choose sharding over vertical scaling (stronger hardware)? Describe three evaluation criteria you would use during a discovery session with stakeholders and the kind of data or metrics you'd request to decide.

EasyTechnical

80 practiced

Explain how partitioning affects transactional guarantees. For a system where some workflows require multi-row updates across shards, outline three approaches to preserve atomicity or achieve acceptable correctness and discuss their trade-offs.

Sample Answer

Partitioning (sharding) breaks the single-node transactional boundary, so multi-row updates that span shards cannot rely on a single local ACID transaction. You must choose between preserving strong atomicity (costly) or accepting weaker guarantees with compensating controls. Three common approaches:

1) Distributed transactions (e.g., 2PC or coordinator+consensus)- What: Use a global coordinator that runs a two-phase commit or a consensus-backed transaction across shards.- Benefits: Preserves ACID semantics across shards.- Trade-offs: High latency, reduced throughput, complex failure/recovery, hard to scale; operationally heavy and can create coordinator bottlenecks.

2) Saga / compensating transactions with orchestration or choreography- What: Break the workflow into ordered local transactions per shard; if a later step fails, run compensating actions to undo prior steps.- Benefits: Scales well, simpler per-shard logic, tolerant of partial failures; fits long-running business flows.- Trade-offs: Only eventual consistency; requires well-defined compensations and idempotency; complexity shifts to business logic and testing; possible temporary inconsistent states.

3) Design for single-shard semantics / co-location / transactional outbox + CDC- What A: Re-design data model so frequently-updated items are co-located (shard by aggregate ID) to keep updates local.- What B: If unavoidable, perform a single-shard transaction to write intent to an outbox and use CDC or a reliable message relay to apply changes on other shards (idempotent consumers).- Benefits: Maintains high throughput and low latency for common cases; outbox+CDC gives reliable cross-shard propagation without 2PC.- Trade-offs: Requires schema/partitioning changes or extra components (CDC, queues); eventual consistency for cross-shard state; operational complexity for outbox and replay handling.

Which to pick:- If business requires strict atomicity (financial settlements), prefer distributed transactions or move data so strong guarantees are local.- If acceptable to be eventually consistent (orders/inventory), use Sagas or outbox+CDC with careful compensations and idempotency.- Always model failure modes, design idempotent operations, provide monitoring and reconciliation jobs to detect and repair inconsistencies.

MediumTechnical

96 practiced

Given these requirements, describe how you would handle cross-shard joins differently for analytical reporting versus low-latency transactional queries: 1) Interactive analytics dashboards (minutes), 2) Order checkout flow (sub-second). Explain your architecture choices for both.

Sample Answer

First, clarify constraints: analytics can tolerate minutes of latency and read-large sets across shards; checkout needs sub-second, strongly consistent joins for a single user/order.

Analytics (interactive dashboards, minutes):- Approach: ETL/ELT into an analytical store where cross-shard joins are pre-computed or executed by a columnar engine.- Architecture: - CDC pipeline (Debezium/Kafka) streams changes from OLTP shards to a data lake/warehouse (Snowflake/BigQuery/ClickHouse). - Apply transformations to produce denormalized tables / star-schema facts and dimension tables, plus materialized views that join across original shards. - Use a scalable query engine supporting distributed joins and late-arriving data; allow eventual consistency and stale results (minutes).- Why: Moves expensive joins off the critical path, leverages batch/stream processing for scale, simpler indexing for analytics.

Transactional (order checkout, sub-second):- Approach: Ensure joins required by checkout are local to a single shard or solved with fast, strongly consistent reads—avoid cross-shard runtime joins.- Architecture: - Design shard key to co-locate customer + cart + inventory partitioning (e.g., customer_id or region) so most joins are single-shard. - For unavoidable cross-shard data (global inventory), maintain strongly-consistent replicated service or cache: - Synchronously replicate critical inventory counters to a small number of leader nodes or use a central fast service (Redis with primary for counters) with CAS/optimistic locking. - Use short-lived distributed locks or implement a Saga pattern for multi-shard commit to avoid two-phase commit latency. - Read-through cache (Redis) for product metadata; fallback to local shard for authoritative writes.- Why: Co-location keeps latency low and atomic; when multi-shard is necessary, prefer eventual approaches bounded by compensating transactions to preserve sub-second user experience.

Trade-offs:- Analytics: higher data duplication and storage cost for speed and simpler queries.- Transactions: increased complexity in sharding strategy and replication; trade simplicity for latency and consistency.

Operational concerns:- Monitor CDC lag, cache hit rates, and tail latency.- Automate resharding plans and have metrics-driven thresholds for when to denormalize more data for transaction paths.

EasyTechnical

75 practiced

List the most important operational metrics and alerts you would implement to monitor the health of a sharded database cluster (think capacity, performance, balance). For each metric explain the reason for monitoring it and a sensible threshold or alerting strategy.

Sample Answer

Below are the key operational metrics and sensible alerting strategies I’d implement to monitor a sharded database cluster, focused on capacity, performance, and balance. For each metric I give why it matters and an alerting approach (thresholds are examples—tune to your workload).

1. Disk usage per shard (GB and %)- Why: full disks cause write failures and node evictions; uneven fill indicates skew.- Alerting: Warning at >70% used; Critical at >90%. Also alert if any shard grows >10% in 1h (rate-of-change). Runbook: throttle writes, add capacity, rebalance chunks.

2. CPU utilization per shard/node- Why: sustained high CPU correlates with latency and may indicate hot shard or bad queries.- Alerting: Warning at 70% for >5m; Critical at 90% for >2m. Correlate with query rates and P99 latency.

3. Memory usage / page cache hit ratio- Why: low cache hit -> higher I/O and latency.- Alerting: Warning when free memory <20% or cache hit <90% sustained 10m; Critical if cache hit <80% or OOM events. Runbook: tune cache, add memory, investigate queries.

4. Disk I/O (I/O wait, read/write latency, queue depth)- Why: I/O bottlenecks cause slow writes/reads across shards.- Alerting: I/O wait >20% for 5m or avg disk latency >10ms (reads) / >20ms (writes) => Warning; double thresholds for Critical. Consider storage scaling.

5. Request/ops throughput and P95/P99 latency per shard- Why: user-visible performance; helps find hot shards.- Alerting: P99 > SLA (e.g., 300ms) for 5m => Warning; >2x SLA => Critical. Also spike detection relative to baseline.

6. Replica/replication lag (seconds) per replica- Why: prevents stale reads, affects failover safety.- Alerting: Warning at >5s; Critical at >30s or sustained growth. If lag spikes, investigate network/IO.

7. Shard key distribution / data size variance- Why: imbalance causes some shards to be overloaded while others are idle.- Alerting: If max shard size > 2x median (or coefficient of variation >0.5) => Warning; >3x => Critical. Runbook: trigger chunk migration or resharding plan.

8. Number and rate of chunk migrations / rebalances- Why: excessive migrations indicate churn, can increase latency.- Alerting: Unexpected sustained migration rate > baseline (e.g., >10/hr) or failed migrations => Warning/Critical. Investigate indexing/ingestion patterns.

9. Connection counts and connection queue length (per mongos / router)- Why: exhausted connections degrade availability.- Alerting: Connections >80% of connection_limit => Warning; queued connections >0 sustained => Critical. Scale routers/pools.

10. Failures / error rates (write errors, timeouts, node restarts)- Why: direct indicator of health and potential data loss.- Alerting: Any node restart => Immediate alert. Error rate >1% of ops or sudden spike => Critical. Include aggregation/alert grouping to avoid noise.

11. Hot-key / hotspot metric (top keys by ops)- Why: identifies single-key overload that requires application or schema change.- Alerting: Top key >10% of ops or sustained high latency correlated to a single key => Warning/Critical. Runbook: introduce sharding with different key, cache, or fan-out.

12. Availability / node health (heartbeat, process up)- Why: detect node loss for failover automation.- Alerting: Node down or heartbeat missed => Immediate Critical with runbook to failover and notify on-call.

Alerting strategy principles:- Use tiered alerts (Warning -> Action, Critical -> Pager).- Require sustained condition (e.g., 3–5 minutes) to reduce flapping.- Correlate related signals (e.g., high latency + high CPU) before paging where possible.- Include contextual tags (shard id, cluster, recent migration) in alerts and link to runbook steps: check slow queries, check recent migrations, add capacity, trigger rebalance, or open incident.

These metrics give capacity (disk/memory), performance (CPU, I/O, latency), and balance (shard size variance, migrations, hot keys) coverage so you can detect root causes quickly and guide remediation.

HardSystem Design

80 practiced

Design a multi-tenant sharding model for a SaaS product where some tenants generate orders of magnitude more load than others. Explain tenant isolation, resource allocation (CPU/disk), cost attribution, and how you would prevent noisy neighbors from affecting small tenants.

Sample Answer

Requirements & constraints:- Multi-tenant SaaS with highly-skewed tenant load (some tenants 100x+ others).- Goals: isolation (stability/security), predictable performance, fair cost attribution, operational simplicity and reasonable cost.

High-level approach — hybrid sharding + tenancy tiers:- Tiered model: small tenants on a pooled/shared tier; large/high-SLA tenants get dedicated shards/instances.- Shard = logical group with its own database instance (or primary + replicas), app node pool, and resource quotas.- Placement decisions use tenant metadata (RPS, storage, SLA) and automated telemetry to move tenants between tiers.

Tenant isolation & resource allocation:- Compute: run app in Kubernetes with Namespaces per shard. Use node pools (spot/standard) and taints/tolerations. Per-pod CPU/memory requests & limits; cgroups enforced on nodes. Large tenants get dedicated node pools or dedicated instance groups.- Disk/IO: put noisy tenants on separate tablespaces / volumes. Use block-level QoS (IOPS limits) at cloud provider (EBS/PD) or storage classes in k8s. For RDBMS, use separate instances for heavy tenants; for shared DBs use resource groups/statement-level throttling.- Network: egress/ingress rate-limiting (API gateway + per-tenant quotas).- Work queues: separate queues per shard with rate controls; background jobs use concurrency limits per tenant.- Cache: per-tenant cache keys with eviction policies; large tenants can get dedicated cache clusters or namespaces (Redis ACLs & VM isolation).

Prevent noisy neighbors:- Admission control: API gateway enforces token-bucket per-tenant RPS/throughput; backpressure via 429s + tiered retry guidance.- Adaptive throttling: dynamic limits based on real-time SLO adherence; lower-priority tenants get reduced capacity when a shard approaches capacity.- Circuit breakers & graceful degradation: disable non-critical features for impacted tenants; serve read-fallbacks from stale replicas.- Dedicated lanes: for top N noisy tenants, reserve dedicated app/db resources.- SLO-based autoscaling: autoscale shard node pools, but cap per-tenant growth to avoid cost explosion.

Cost attribution & billing:- Metric collection: tag all telemetry with tenant_id (API, app, DB, storage, network). Aggregate usage: CPU-seconds, memory-hours, disk-GB-month, IOPS, network bytes, cache ops, background-job-seconds.- Cost model: combine fixed tier fee + usage-based invoicing for excess (e.g., per-GB, per-1000 requests, per-CPU-hour). For dedicated resources, pass-through full resource costs.- Billing pipeline: export metrics to analytics (Prometheus -> Thanos/Influx + billing service) and map costs using cloud unit prices; produce chargeback & show in tenant portal.- Chargebacks for noisy behavior: automated alerts and billing for sustained high I/O or spikes; thresholds & negotiated rates in SLOs.

Operational & scaling considerations:- Automated placement service that observes telemetry and recommends/migrates tenants (live migrations for stateless, controlled data migration for DB).- Observability: per-tenant dashboards, SLA/SLO tracking, alerts for noisy patterns.- Security: tenancy isolation via authz, encryption, network policies.- Trade-offs: - Dedicated instances increase cost but provide strong isolation; pooling is cheaper but needs enforcement mechanisms. - Aggressive throttling protects stability but can hurt revenue/UX—mitigate via clear SLAs and upgrade paths.

Example flow:- New tenant defaults to pooled tier. After monitoring shows sustained >X RPS or >Y IOPS, placement service schedules migration: create dedicated DB instance, provision node pool, update routing rules, migrate data, update billing. During surge, gateway throttles except for dedicated tenants; autoscaling adds nodes up to configured caps.

This design balances cost-efficiency for small tenants with strong isolation and predictable performance for large ones, while providing clear attribution and operational controls to prevent noisy neighbors.

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.