Distributed SQL and Query Scaling Questions
Principles and practices for running and optimizing SQL queries in distributed query engines and cloud data warehouses. Candidates should understand how distributed execution affects query performance including partitioning strategies, shuffle operations, data skew, partition pruning, and cost based optimization in engines such as Spark SQL, Presto, and BigQuery. This topic includes designing queries to minimize data movement, choosing appropriate partition keys, leveraging cluster resources efficiently, and interpreting execution plans and job stages to diagnose bottlenecks in large scale queries.
MediumTechnical
55 practiced
A Spark job preparing training data is spilling often to disk during shuffle and failing memory checks. Describe a step-by-step debugging and remediation plan: what Spark UI metrics you check, what config flags you adjust, and code changes you might make to reduce spills and improve stability.
Sample Answer
Situation: A Spark job preparing ML training data is spilling during shuffle and failing memory checks. I’d follow a diagnostic -> remediation loop: verify metrics, change configs, and apply code fixes.1) Diagnose via Spark UI & logs- Stages/Tasks: identify which stage(s) have high Shuffle Read/Write and long task times.- Executors tab: check failed tasks, executor memory used, peak JVM heap, GC time.- Stage details: “Shuffle Read/Write Size”, “Shuffle Records Read/Write”, and “Memory Spill (map/reduce)”.- Environment & Executors: spark.memoryManager/tungsten metrics, off-heap usage.- Driver/executor logs: look for OutOfMemoryError, repeated GC, and “spill” messages.- Metrics: spark.sql.shuffle.partitions vs task time variance (skew), bytesRead/bytesWritten per task.2) Quick config adjustments (safe, incremental)- Increase executor memory &/or cores per executor: spark.executor.memory, spark.executor.cores (watch GC).- Increase shuffle buffer and reduce network waits: spark.shuffle.file.buffer (e.g., 32k→64k), spark.reducer.maxSizeInFlight (default 48MB -> adjust).- Use sort-based shuffle (default): spark.shuffle.manager=sort- Enable/compress shuffle: spark.shuffle.compress=true, spark.shuffle.spill.compress=true- Tune unified memory: spark.memory.fraction (default 0.6) and spark.memory.storageFraction (default 0.5) if jobs are compute- or cache-heavy.- Increase parallelism to reduce per-task data: spark.sql.shuffle.partitions and spark.default.parallelism.- Enable off-heap if needed: spark.memory.offHeap.enabled=true and spark.memory.offHeap.size3) Code-level remediations- Reduce shuffle volume: - Push filters and selects earlier (prune columns, filter rows before joins/aggregations). - Convert wide records to compact types (avoid strings; use ints/enums).- Repartition strategically: - Use repartition(n) to increase partitions before heavy shuffles or coalesce after high parallelism; ensure n ≈ total cores * 2-4. - Use partitionBy on stable keys for repeated operations.- Reduce skew: - Identify hot keys via sampled counts; apply salting or custom partitioner or use approximate algorithms.- Map-side aggregation: - Use reduceByKey/aggregateByKey/ map-side combine (instead of groupByKey) to reduce shuffle bytes.- Broadcast small tables: - Use broadcast join for small lookup tables (spark.sql.autoBroadcastJoinThreshold or broadcast()).- Cache judiciously: - Persist only what’s reused, and use MEMORY_AND_DISK to avoid failures.- Serialization: - Use Kryo (spark.serializer=org.apache.spark.serializer.KryoSerializer) and register classes.4) Monitoring & iterative tuning- After each change, re-run a representative job, watch Spark UI for spills, GC and task durations.- If GC dominates, lower executor cores (1-3) to reduce concurrent tasks per JVM and decrease memory pressure.- If still failing, collect heap/GC logs and consider larger instance types or scaling out.Trade-offs & rationale:- Increasing memory reduces spills but costs more; increasing partitions reduces per-task memory but increases overhead—balance based on job profile.- Map-side aggregation and broadcasting reduce network I/O (best for ML ETL), while repartitioning/salting handle skew.Result expectation:- Apply the smallest effective config change first (increase partitions, push filters), then memory or code changes. Aim to eliminate shuffle spills and bring GC under ~20% of task time; repeat until stable.
HardTechnical
63 practiced
Discuss how inaccurate or stale statistics (NDV, histograms) can lead to suboptimal plans in distributed SQL engines. For an append-only streaming dataset, propose practical mechanisms to maintain effective statistics that balance collection cost and optimizer accuracy.
Sample Answer
Definition / why it matters:Inaccurate or stale statistics (NDV, histograms) cause the optimizer to misestimate cardinalities and value distributions. In distributed SQL engines that decide join order, partitioning, and parallelism, bad estimates can yield suboptimal plans: wrong join ordering (nested-loop vs hash), skewed partitioning causing stragglers, under/over-provisioned parallelism, and expensive network shuffles. For append‑only streaming data these errors compound because data distributions drift over time.Practical mechanisms (mix of streaming sketches, selective sampling, and adaptive refresh):1. Mergeable sketches for NDV and quantiles- Use HyperLogLog (HLL) for per-partition NDV and t-digest or DDSketch for approximate histograms/quantiles. They are mergeable, memory‑bounded, and cheap to update on each batch. Accuracy vs size is tunable.2. Reservoir & stratified sampling- Maintain reservoir samples per key-sketch strata (hot keys, tail) so the optimizer can detect skew. Update periodically; keep separate small reservoirs for recent window to capture drift.3. Windowed / weighted summaries- Keep time-decayed sketches (exponential decay or sliding windows) so recent data has higher influence—important for append-only streams with drift.4. Hybrid full-refresh policy- Full histogram rebuilds on low-load periods or when statistical divergence triggers exceed thresholds (see monitoring). Cheap sketches run continuously; expensive full stats run infrequently.5. Adaptive triggers & feedback- Monitor query plan performance (actual vs estimated cardinalities, executor skew). If estimation error exceeds thresholds, trigger targeted re-sampling/re-sketching of affected columns or partitions.6. Prioritization- Focus heavy-weight stats on columns used in joins, group-bys, and filters. Less critical columns get only sketches.7. Store and expose confidence metadata- Attach error bounds and timestamps to stats so optimizer uses robust plans (e.g., choose join reorderings conservatively when confidence low).Operational considerations and trade-offs:- Cost: sketches + periodic sampling are low overhead; full histograms are costly but can be scheduled.- Accuracy: t-digest + HLL provide high practical accuracy; stratified sampling reduces skew risk.- Complexity: requires integration between streaming ingestion, stats service, and optimizer; but mergeability simplifies distributed aggregation.Metrics to monitor:- Query cardinality estimation error distribution, shuffle sizes, task skew, and re-plan frequency. Use these to tune sketch sizes, sampling rates, and refresh policies.This combined approach balances continuous low-cost accuracy with targeted expensive refreshes, minimizing plan regressions while keeping collection overhead acceptable.
MediumTechnical
80 practiced
Explain how executor count, cores per executor, and executor memory influence parallelism and GC behavior for large Spark SQL jobs that prepare training datasets. Provide a concrete example configuration and reasoning for a cluster tasked with heavy aggregations and large shuffles.
Sample Answer
Start with the fundamentals: Spark parallelism is driven by number of tasks, which is bounded by number of partitions. How those tasks are executed depends on how you break up cluster CPU and memory into executors.- Executor count (number of JVMs) controls parallelism granularity and isolation. More executors → more JVMs → more independent GC heaps (smaller heaps each) and more task slots overall (if cores per executor balanced).- Cores per executor determines concurrent tasks per JVM. Fewer cores per executor (2–4) gives better GC behavior because each JVM has fewer concurrent allocations and shorter GC pause surfaces; more cores (≥5–8) can increase throughput but risks long, stop-the-world GC pauses and greater GC fragmentation during heavy aggregations.- Executor memory sets heap size for in-memory data, aggregation buffers, and shuffle blocks. Too small → frequent spills to disk (slow), too large → long GC pauses. Also remember JVM overhead (spark.executor.memory + spark.yarn.executor.memoryOverhead).For heavy aggregations and large shuffles: aim for more executors with moderate cores to maximize parallelism and keep heaps moderate to limit GC pauses. Also tune shuffle partition count to match parallelism.Concrete example (cluster with 100 worker cores, 800 GB total RAM):- Target per-executor cores: 4- Executor count = floor(100 / 4) = 25 executors- Per-executor memory (heap): ~27 GB (set spark.executor.memory=27g) and memoryOverhead=3g → 30 GB container- spark.executor.cores=4- spark.sql.shuffle.partitions = 25 * 4 * 2 = 200 (2× total cores is a good starting point)- GC flags: spark.executor.extraJavaOptions = "-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:InitiatingHeapOccupancyPercent=35 -XX:+UseStringDeduplication" (Java 11+)- spark.memory.fraction = 0.6 (default 0.6) and spark.memory.storageFraction = 0.3 — adjust if caching large feature tables.Reasoning:- 4 cores/executor gives 25 JVMs; each JVM has moderate heap (27 GB) — reduces maximum pause time vs fewer big JVMs.- 200 shuffle partitions spreads shuffle/merge work across all cores and avoids huge partitions; more partitions means smaller tasks and better parallelism but overhead to schedule many tiny tasks.- G1GC works better than CMS for large heaps and reduces long full-GC events; tuning MaxGCPauseMillis biases for latency.- Having many executors reduces blast radius of a single GC pause; if you used 1 executor with 100 cores and 800GB, a single GC pause would stall many tasks.Other practical tips:- If you observe frequent spills: increase executor memory or increase shuffle.partitions so each partition is smaller.- For skewed keys: apply salting or custom partitioner, or increase numReducers for problematic stages.- Use dynamic allocation with min/max executors if workload varies, but set conservative min to avoid excessive small executors during heavy shuffles.- Monitor GC logs, shuffle read/write sizes, task times; iterate: if GC dominates, reduce heap or cores per executor; if I/O dominates, increase memory and reduce spills.This configuration balances parallelism (many tasks running) with manageable JVM heap sizes to limit GC impact while providing enough memory to reduce spills during large aggregations and shuffles.
HardSystem Design
69 practiced
Design a system to provide point-in-time correct joins for model training at petabyte scale. Describe how you'd store versioned features, support time travel lookups, perform efficient backfills, and avoid joining the entire event history for each model training run.
Sample Answer
Requirements:- Point-in-time correctness: for each training example (entity_id, label_time) return feature values as-of label_time (no leakage).- Petabyte scale, low-latency lookups for training and serving, efficient backfills, reproducible versions.High-level design:1. Storage & versioning- Use a columnar, ACID lake (Delta Lake / Apache Iceberg) as canonical feature store. Store features as append-only tables keyed by (entity_id, feature_name, event_timestamp, ingestion_version).- Leverage the table transaction log for time-travel (Delta time-travel / snapshot isolation). Each commit has a version id enabling reproducible reads.2. Data model & partitioning- Primary key: entity_id. Each feature row has event_timestamp and source_version.- Partition by hashed entity_id ranges and date-buckets of event_timestamp to limit scan scope. Cluster-sort by (entity_id, event_timestamp desc) so latest-per-entity before cutoff is fast.3. Point-in-time lookup API- Training pipeline provides a label table with (entity_id, label_time, example_id).- Implement a distributed join service that, for each example, issues a "latest-before" lookup: SELECT first_row_per_entity WHERE event_timestamp <= label_time AND commit_version <= training_snapshot. Use window function row_number() over (partition by entity_id order by event_timestamp desc) filtered to row_number=1.- Use Delta snapshot version corresponding to training run to guarantee reproducibility.4. Avoid joining full history- Precompute materialized “latest-as-of” feature views per coarse time-slice (e.g., daily snapshots of "latest per entity") for heavy features. For arbitrary label_time, read nearest snapshot plus incremental events between snapshot_time and label_time (small delta).- Maintain compacted "state" tables: periodically compact event streams into entity-state rows (one row per entity per snapshot) via compaction jobs. Training joins snapshot instead of full event history.5. Efficient backfills- Backfill orchestrator picks a target snapshot (version) and processes label ranges in parallel by entity partitions. Use incremental backfill: if materialized snapshots exist, only recompute deltas.- Use vectorized engine (Spark) with map-side data pruning using partition/zone-maps and bloom filters to avoid scanning irrelevant files.6. Streaming/online ingestion & CDC- Ingest feature updates via CDC/streaming into Delta with monotonically increasing ingestion_version. Streaming allows building real-time incremental snapshots.7. Scalability & optimizations- Use file-level statistics, min/max per column and Z-ordering to prune files.- Use join strategies: map-side broadcasts for small feature sets, partitioned joins for large.- Use Bloom filters (per-file) and pushdown predicates event_timestamp <= label_time to limit reads.- Cache hot partitions and pre-warm materialized views for frequent label_time ranges.8. Reproducibility & governance- Record training_snapshot_version, feature-table versions, transformations (as code) in metadata registry. Provide deterministic replay by restoring snapshot versions.Trade-offs:- Frequent compaction increases write cost but reduces read cost (training). Snapshot cadence balances storage vs backfill cost.- Materialized snapshots use space but dramatically speed up training/backfills.This design yields point-in-time correctness via snapshot isolation and latest-before semantics, avoids scanning entire event histories through compaction/materialized snapshots and partitioned lookups, and supports efficient backfills with incremental recomputation and parallelization.
EasyTechnical
63 practiced
Explain the difference between table partitioning and bucketing in distributed SQL engines (e.g., Spark SQL, Presto, BigQuery). In the context of ML feature tables used for daily model training (hundreds of GBs to TBs), describe when you would choose partitioning, when bucketing helps, and how each affects join performance, shuffle volume, and file layout.
Sample Answer
Partitioning and bucketing both organize data to reduce I/O and shuffle, but they work differently and suit different ML feature-table patterns.What they are- Partitioning: splits table into directory-level slices by a partition key (date, hour, region). Engines prune partitions during reads.- Bucketing: hashes a column into a fixed number of files/buckets (e.g., 64 buckets). Bucket id is stable so rows with same key land in same file across tables if bucket count/algorithm match.When to use for daily ML feature tables- Partition by ingestion time or training window (date/hour). For daily training on hundreds of GB–TB, partitioning by date massively reduces scanned data and speeds snapshot/extracts.- Use bucketing when you frequently join large feature tables on a stable key (user_id, entity_id) and can control bucket count. If both tables are bucketed compatibly, joins can be map-side (no full shuffle) or require much less shuffle.Effects on join performance, shuffle, file layout- Partitioning: reduces scan cost; does not by itself reduce shuffle for joins unless partition key = join key and engine can colocate partitions. File layout: many directories (one per partition); could create many small files if not managed.- Bucketing: improves join performance when bucketed on join key and bucket counts align—can avoid full shuffle or use smaller shuffle by bucket. File layout: fixed number of files per table (good for parallelism), helps avoid small-file explosion if buckets sized well.Practical notes- Combine both: partition by date and bucket by join key within partitions for best of both worlds.- Choose bucket count = expected parallelism (executor cores) and keep stable; rebucketing is expensive.- Maintain compaction and vacuuming to avoid too many small files; monitor skew (hot keys) which hurts bucket balance.
Unlock Full Question Bank
Get access to hundreds of Distributed SQL and Query Scaling interview questions and detailed answers.
Sign in to ContinueJoin thousands of developers preparing for their dream job.