Distributed Query Execution, Shuffle, And Skew

What's being tested

Interviewers probe your practical understanding of distributed query execution: how shuffle moves data between stages, why skew breaks parallelism, and which execution strategies (partitioning, join algorithms, spilling, broadcasting, and adaptive execution) reduce latency and cost. Databricks cares because efficient shuffles directly affect job tail latency, cluster utilization, and SLOs; you must show concrete tradeoffs, measurable metrics, and an actionable mitigation plan a backend engineer would implement or tune.

Core knowledge

Shuffle: the network+disk phase that redistributes records by partition key between map and reduce tasks; cost dominated by bytes written, transferred, and read. Metrics: shuffle write bytes, shuffle read bytes, and shuffle spill.
Partitioning strategies: hash partitioning (fast, uniform if keys random) vs range partitioning (ordered, good for range queries) vs custom partitioners; average per-partition size = total_size / num_partitions.
Skew definition & metric: skew factor = max_partition_size / avg_partition_size; values ≫1 indicate bad skew. Tail latency often driven by single largest partition (straggler).
Join algorithms: broadcast join (send small table to all workers, cost ~size_small * num_workers), shuffle hash join (hash & exchange both sides), sort-merge join (sort partitions then merge; good for large inputs and range-partitioned keys).
Thresholds & heuristics: broadcasting practical when the small side is ≲ tens of MBs per worker (Spark default ~10–100 MB); otherwise prefer shuffle join. When per-partition memory > available executor RAM, expect spill-to-disk and higher latency.
Spill and external shuffle: when in-memory buffers exceed memory, tasks spill to disk; external shuffle services (e.g., Spark External Shuffle Service) decouple shuffle file lifetime from executors but add I/O and management complexity.
Skew mitigation techniques: salting (prefix keys with random salt to spread hot keys), partial aggregation (pre-aggregate on mappers), skew-aware joins (detect heavy keys and handle separately), and increasing partitions (finer parallelism but higher overhead).
Adaptive Query Execution (AQE): runtime adjustments based on observed metrics — e.g., dynamically change number of reducers, convert shuffle join to broadcast join when small, or split skewed partition into multiple tasks.
I/O and network tradeoffs: increasing partitions reduces per-task memory but raises metadata and RPC overhead; salting increases shuffle volume by factor ≈ average_salt_count.
Speculation & retries: enable speculative execution to mitigate transient stragglers, but beware job duplication amplifying load on hot partitions.
Instrumentation to act: know where to read shuffleReadMetrics, shuffleWriteMetrics, executor logs, per-task CPU/wall time, and OS-level metrics (disk IO, network bandwidth) to attribute bottlenecks.
Complexity and cost model: model job time ≈ max_over_partitions(read + compute + write + network) where network ≈ shuffle_bytes / network_bandwidth; optimizations should reduce either bytes or critical path.

Worked example — "Explain how shuffle works and how to mitigate skew"

First 30s: ask clarifying Qs — input sizes per relation, key cardinality and distribution (Zipfian?), available executor memory, and whether joins are equi-joins or range joins. Frame goal: minimize job p99 latency and network I/O.

Answer skeleton:

Describe shuffle phases: partitioning at mappers, writing local files, block transfer, and reduce-side read/merge.
Show cost model: per-partition bytes and latency dominated by largest partition; compute skew factor.
Present mitigation options in order: broadcast if small, pre-aggregate, increase partitions, salting or skew-aware split, AQE, and speculative execution.
Implementation notes: tune spark.sql.shuffle.partitions, broadcast threshold, and monitor shuffleReadBytes/shuffleWriteBytes.

Tradeoff to call out: salting reduces straggler risk but multiplies shuffle bytes and complicates correctness (requires un-salting post-aggregation). Communicate measurable criteria: apply salting only when skew factor > X (e.g., 10) and heavy-key bytes exceed single-task memory.

Close: propose an experiment plan — run instrumented job with varying spark.sql.shuffle.partitions, enable AQE, and compare p50/p95/p99 and shuffle bytes; say "if more time, I'd prototype salting and measure end-to-end cost vs benefit."

A second angle — "Detecting and fixing skew at runtime on production jobs"

Frame: you cannot change job code easily; you can observe metrics and apply runtime fixes. Strategy: detect hot partitions via shuffleReadMetrics and task durations, then (1) if small table detected, trigger broadcast dynamically; (2) if skewed keys present, use AQE to split large partitions into multiple reducers; (3) enable speculative tasks for the straggler. Emphasize low-risk fixes first (tuning partition count, enabling AQE/speculation) before code-intrusive solutions (salting). Highlight tradeoffs: AQE relies on accurate sampling and may not catch transient spikes; speculative execution duplicates work and can worsen load if the system is I/O-bound.

Common pitfalls

Pitfall: assuming uniform key distribution — candidates often propose hash partitioning without checking for Zipfian/long-tail distributions; always inspect key cardinality and frequency histogram.

Pitfall: recommending "just increase partitions" without cost modeling — more partitions can reduce per-task memory but increases small-file, metadata, and network overhead, sometimes worsening latency.

Pitfall: suggesting broadcast join blindly — if the broadcasted dataset exceeds memory or the cluster has many executors, broadcasting increases GC pressure and may OOM workers; prefer runtime checks and AQE to flip strategies.

Connections

Query optimizers and cost-based optimization (how statistics enable broadcast vs shuffle decisions).
Task scheduling and speculative execution, because skew creates stragglers that schedulers may mitigate or exacerbate.
Storage and I/O systems (local SSDs, remote shuffle service) since disk throughput and latency often dominate spilled-shuffle cost.