Data Engineering Interview: How Do You Diagnose and Fix Spark Data Skew?

Prompt and When It Applies

A daily Spark SQL job left-joins a 4.8 TB event fact table to a 180 GB product dimension on product_id. After an upstream release, 32% of events are normalized to product_id='UNKNOWN'. The job uses 2,000 shuffle partitions. In the join stage, the Spark UI shows median task shuffle read of 1.1 GiB, while one task reads 720 GiB, spills repeatedly to disk, and eventually fails with OOM after retries. Runtime has grown from 24 to 96 minutes. The business needs the job to finish within 45 minutes, without discarding unknown-product events or changing the left-join result.

The table sizes, key share, partition metrics, runtimes, and SLA are interview assumptions. The central task is to distinguish data skew from insufficient resources and join-output explosion using partition-level evidence, then choose a remedy that preserves the data contract. This belongs in the data category because it tests Spark execution plans, shuffle partitions, data distribution, and batch correctness. The existing Kafka hot-partition question focuses on message keys, ordering, and consumer offsets. This question focuses on runtime SQL partitions, join strategy, AQE, and result conservation, so the failure layer and validation method are different.

The same reasoning applies to groupBy, distinct, window functions, and other wide dependencies. When many records for one key converge on a few post-shuffle tasks, they can create stragglers, spill, GC pressure, or OOM. A good answer does not begin by adding memory. It first proves whether the slowest task owns a disproportionate amount of data and computation.

What the Interviewer Evaluates

Start with evidence granularity. A strong answer moves from the job to the SQL query, then to a specific stage and its individual tasks. It compares duration, shuffle-read records and bytes, spill, peak execution memory, and GC time. A task that reads hundreds of times the median and remains slow when retried on another executor supports deterministic data skew. Aggregate executor memory and total runtime alone do not establish that cause.

Execution-plan literacy matters just as much. EXPLAIN FORMATTED confirms the Exchange, join type, and physical join. EXPLAIN COST and runtime statistics in the SQL UI expose estimated and observed data sizes. Spark AQE uses runtime statistics to adjust the plan. In Spark 4.2.0, skew-join optimization can split a skewed sort-merge-join partition and replicate the smaller side when required. The documented defaults mark a partition skewed only when it exceeds both five times the median and 256 MiB. A candidate should inspect the effective environment configuration because documentation defaults are not immutable cluster facts.

Data semantics become the dividing line. UNKNOWN may be a valid “unattributed” event or an upstream defect. Dropping, randomly distributing, or rewriting those rows can change the result. A salted join must replicate only hot-key rows from the dimension and assign a deterministic salt to hot fact rows. Replicating the whole dimension multiplies data volume, while independently randomizing both sides loses matches.

Remedy selection reveals another level of depth. Raising spark.sql.shuffle.partitions creates more hash buckets, but every row for one hot key still lands in one bucket. Broadcast is suitable only when projection, filtering, and reliable statistics prove that one side safely fits on every executor; forcing a 180 GB dimension to broadcast is unsafe. AQE is the low-intrusion first choice. Explicit salting fits stable hot keys when AQE does not trigger or still misses the SLA. Aggregations often use a salted partial aggregation followed by a second merge.

Complete validation closes the answer. A performance fix must also prove that row counts, business amounts, unknown keys, unmatched rates, and duplicate rates are unchanged. Reducing runtime from 96 to 40 minutes does not prove correctness or show whether the solution survives a different key distribution tomorrow.

Questions to Clarify Before Answering

• Is the bottleneck in scan, shuffle write, or after shuffle read? Uneven scan tasks may come from huge or unsplittable files. This prompt reaches the key-skew path because the outlier appears after the join shuffle.
• Does 32% refer to records, compressed bytes, or processing cost? Wide rows, expensive UDFs, and output fan-out can create cost skew even when row counts look modest. Compare records, bytes, time, and output rows.
• What does UNKNOWN mean to the business? If unknown products need no dimension attributes, split them from the main join and fill null attributes under the original contract. If they must match a sentinel dimension row, preserve the join and split the hotspot.
• Is product_id unique in the product dimension? Several UNKNOWN dimension rows turn the hot fact key into a many-to-many output explosion. Separate bad cardinality from partition skew before tuning.
• Which Spark version and AQE settings are effective? Check the Environment page and final adaptive plan for switches, thresholds, join type, and evidence that skew splitting actually ran.
• Is 180 GB the original dimension or its projected join input? If filtering to the key and two attributes makes it safely broadcastable, broadcast may beat a two-sided shuffle. Runtime statistics and executor-memory budgets must prove that case.
• Which invariants define an equivalent result? At minimum, specify total rows, unique events, additive business measures, unknown-key rows, unmatched rows, and permitted duplicate semantics.
• Are hot keys stable and enumerable? A few stable keys fit targeted salting. A changing long tail favors AQE, dynamic hot-key detection, or an upstream semantic correction.

30-Second Answer Framework

“I would compare join-task shuffle read, spill, GC, and retry location in the SQL UI. A 720 GiB task against a 1.1 GiB median, plus 32% UNKNOWN, supports key skew; I would also verify dimension uniqueness to exclude output explosion. First I would confirm that AQE skew join splits the large partition. If runtime still exceeds 45 minutes, I would salt UNKNOWN deterministically by stable event_id and replicate only its sentinel row. Finally, I would reconcile rows and amounts with the baseline, then compare max-to-median task input, stage time, and cost.”

Step-by-Step Deep Answer

Step 1: Attribute the 96 minutes to a stage and task

Retain the event log and use the Spark History Server to compare a normal and regressed run for the same data date. Follow the SQL query into its stage details. Record the percentiles and maximum task duration, aligned with shuffle-read records and bytes, shuffle spill, peak execution memory, GC time, failure reason, and executor. In the SQL plan, check for an Exchange hashpartitioning(product_id, 2000) before the left join, identify the final physical join, and determine whether the adaptive plan completed.

Here, the same task still reads about 720 GiB after retrying on another executor, while most tasks read about 1.1 GiB. That points to the input partition itself. If a slow task has ordinary input and high GC or disk wait on one executor, investigate the node first. If every task spills uniformly, focus on overall partition sizing and resource budgets. If output rows multiply suddenly, inspect dimension duplicates and the join condition.

Step 2: Prove the cause with key distribution and join cardinality

Measure hot keys after applying exactly the same filters and key normalization used by production. Inspecting the raw column is insufficient because trim, case normalization, coalesce, or a UDF may collapse several values into one key. On a very large table, use existing statistics, a controlled sample, or a bounded aggregation so the diagnostic does not become another unconstrained job. The query below assumes that the fact table already contains or precomputes payload_bytes, allowing both row and byte skew to be measured. Without that column, use storage statistics or a controlled serialization estimate. The query expresses the required accounting:

SELECT
  COALESCE(product_id, '<NULL>') AS join_key,
  COUNT(*) AS row_count,
  SUM(payload_bytes) AS payload_bytes
FROM fact_events
WHERE event_date = DATE '2026-07-17'
GROUP BY COALESCE(product_id, '<NULL>')
ORDER BY row_count DESC
LIMIT 20;

Also prove that each product_id appears at most once in the dimension and compare fact-event counts before and after the join. For this left join, a unique dimension key means each fact event produces exactly one row, including unmatched events. If UNKNOWN owns 32% of fact rows, the dimension has one sentinel row, and output does not multiply, hash partitioning explains the single 720 GiB partition.

Step 3: Address the semantic cause before choosing an execution technique

Investigate why the upstream release maps 32% of events to UNKNOWN. If it is a regression, roll back or fix the mapping and rerun affected partitions. That restores both data quality and performance. If it is a valid business value, the execution layer must support the distribution.

Valid unknown events that need no product attributes can take a separate path: join cold keys normally, fill dimension attributes with nulls for the unknown path under the existing contract, and combine with unionByName. That removes a shuffle with no information gain. If UNKNOWN must match sentinel attributes, retain the join and split the work with AQE or targeted salting. Result semantics decide the branch; uniform partitions are not permission to discard data.

Step 4: Choose the least costly remedy that works

Check AQE first. In Spark 4.2.0, spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled default to enabled, but a cluster, job, or managed platform can override them. The skew-join factor and absolute-byte thresholds must both match. Inspect the final adaptive plan for skew handling and confirm that stage metrics show the large partition split. For a compatible sort-merge join, AQE can split the large side and replicate the smaller side. It adapts to changing daily distributions, but may add shuffle and replication cost, and it cannot repair a logically incorrect many-to-many join.

If the projected dimension has reliable statistics and is truly small, evaluate a broadcast hash join so the fact side does not shuffle on the join key. The original 180 GB is far outside an ordinary broadcast budget. A forced hint may exhaust every executor. Judge projected bytes, concurrent tasks, executor heap, and broadcast timeout together.

When AQE does not trigger or still misses the SLA, manually salt stable hot keys. The following code assumes that event_id is stable and unique, product_id is unique in the dimension, and only UNKNOWN needs to be split. Hot fact rows map deterministically to 32 salts. Only the matching sentinel dimension row is replicated 32 times. Cold keys remain on salt 0, so the full dimension is never multiplied.

from pyspark.sql import functions as F

SALT_BUCKETS = 32
HOT_KEYS = ["UNKNOWN"]

events_salted = events.withColumn(
    "salt",
    F.when(
        F.col("product_id").isin(*HOT_KEYS),
        F.pmod(F.xxhash64("event_id"), F.lit(SALT_BUCKETS)).cast("int"),
    ).otherwise(F.lit(0)),
)

salt_values = spark.range(SALT_BUCKETS).select(
    F.col("id").cast("int").alias("salt")
)

products_hot = (
    products.filter(F.col("product_id").isin(*HOT_KEYS))
    .crossJoin(salt_values)
)
products_cold = (
    products.filter(~F.col("product_id").isin(*HOT_KEYS))
    .withColumn("salt", F.lit(0))
)
products_salted = products_cold.unionByName(products_hot)

result = (
    events_salted.join(products_salted, ["product_id", "salt"], "left")
    .drop("salt")
)

Thirty-two is an initial candidate under this interview scenario. Derive the bucket count from hot-partition bytes, target task size, available parallelism, and small-side replication cost, then test on representative data. Too few buckets preserve a long tail. Too many add scheduling, file, and replication overhead. Merely running repartition(4000, "product_id") still places every UNKNOWN row in one partition.

For groupBy(product_id), there is usually no dimension to replicate. First partially aggregate by (product_id, salt), then merge the partial results by product_id. Operations that can be safely decomposed with associative and commutative merges, such as sum, count, min, and max, fit this technique. Exact median, order-dependent aggregation, and non-mergeable UDF state require a different algorithm.

Step 5: Put correctness, performance, and cost in one acceptance gate

Run the baseline and candidate against one immutable input snapshot. Correctness comes first: compare total output rows, unique event_id, UNKNOWN rows, unmatched rows, and business sums and counts by meaningful dimensions. Take row-level diffs for hot keys, cold keys, nulls, and duplicate dimension keys. The guarantee that one fact event produces one left-join row depends on dimension-key uniqueness, so monitor that constraint separately.

For performance, compare p50, p95, and maximum task duration in the join stage, max-to-median shuffle read, spill, GC, OOM, task retries, stage time, and total runtime. For cost, record executor-hours, shuffle bytes, and output-file count. Finishing within 45 minutes is only one gate. A run that meets the SLA by doubling shuffle, changing results, or failing on the next day's new hotspot is not acceptable.

Release by replaying one historical date, then shadowing a new data date and comparing results. Monitor the hot-key set, max-to-median task-input ratio, and unknown-key share. The jump to 32% UNKNOWN after an upstream release should also trigger a data-quality alert, exposing the semantic regression before it delays the job.

High-Quality Sample Answer

“I would first attribute the regression to a specific join stage in the SQL UI. The current evidence strongly suggests skew: across 2,000 tasks, median shuffle read is 1.1 GiB, one task reads 720 GiB, and that task remains slow after moving to another executor. I would inspect the final adaptive plan, spill, and GC, then profile keys after the exact production normalization. I would also assert dimension-key uniqueness. Multiple UNKNOWN dimension rows would mean the symptom includes join-output explosion.

Assuming a unique dimension and 32% UNKNOWN fact rows, one hash-shuffle bucket explains the straggler. More partitions create additional buckets but do not divide that key, and more executor memory only postpones OOM. I would first determine whether the upstream mapping is a regression. If unknown rows need no dimension attributes, I would split them from the join and fill null attributes with the original left-join semantics. If they must match a sentinel row, I would verify that AQE skew join actually appears in the final plan because it can split a skewed sort-merge-join partition using runtime statistics and replicate the small side.

If AQE still leaves runtime above 45 minutes, I would use targeted salting. A stable event_id would assign each UNKNOWN fact row deterministically to, for example, one of 32 salts. I would replicate only the dimension's UNKNOWN row across those 32 salts; every cold key would use salt 0. Each event still matches one dimension row, while several tasks share the hot-key work. I would derive the final bucket count from hotspot bytes and target task size rather than hard-coding 32 without measurement.

For validation, I would compare the same immutable input against the baseline. Total rows, unique events, unknown and unmatched records, and business sums must agree. I would then compare max-to-median shuffle read, the task-duration tail, spill, OOM, stage time, executor-hours, and output files. Finally, I would replay one historical date, shadow one new date, and alert on unknown-key share and new hotspots. That proves the job meets 45 minutes, preserves results, and remains observable when upstream distribution changes again.”

Common Mistakes

• Raising shuffle partitions from 2,000 to 8,000 immediately → one hot key still hashes to one partition while other tasks get smaller → measure key distribution, then split the key with AQE, semantic branching, or targeted salting.
• Only increasing executor memory → it raises one task's tolerance but leaves 720 GiB of work and the long tail intact → reduce the maximum partition workload first, then size resources from measured tasks.
• Declaring skew after seeing one slow task → a bad node, GC, remote fetch, or slow UDF can also create a straggler → compare task input, retry location, spill, GC, and the execution plan.
• Randomly salting the fact and dimension independently → salts fail to match and lose join results, while retries can become nondeterministic → derive fact salt from a stable row ID and enumerate the same salts on the dimension.
• Replicating the full dimension 32 times → a 180 GB dimension creates enormous network and memory cost → replicate only confirmed hot-key rows and keep cold keys on salt 0.
• Forcing the 180 GB dimension to broadcast → every executor must hold the broadcast data and may fail with OOM → project and measure first; broadcast only after memory and concurrency budgets prove it safe.
• Filtering UNKNOWN to make the job fast → output semantics and downstream metrics change → establish the unknown-event contract and preserve left-join results even when branching.
• Comparing only total runtime → an apparent speedup may come from dropped, duplicated, or miscomputed data → prove row and business invariants before comparing task distribution, cost, and SLA.

Follow-Up Questions and How to Answer

AQE is enabled. Why was the skewed join not split?

Inspect the final adaptive plan and effective settings for spark.sql.adaptive.enabled, the skew-join switch, the median-factor threshold, and the absolute-byte threshold. Both thresholds must match. Confirm that the physical join follows a supported AQE path, runtime statistics are available, and a hint or platform override does not constrain the plan. Test threshold changes or forced skew optimization on representative data while measuring extra shuffle. If the plan cannot benefit, use targeted salting. A true value in a configuration file does not prove that the executed plan split the partition.

If projection reduces the dimension to 6 GiB, can you broadcast it?

Six GiB still needs an executor-heap, concurrent-task, serialized-size, broadcast-timeout, and cluster-stability assessment. Broadcast can remove the large side's join-key shuffle and thereby avoid the hotspot, but it also copies the dimension to executors. Use statistics to prove actual bytes, then observe peak memory and GC in a production-shaped load test before adding a hint. “Much smaller than the fact table” is not a sufficient broadcast criterion.

What if the hot keys change every day and HOT_KEYS cannot be maintained manually?

Prefer AQE's runtime response. If explicit salting is still necessary, produce a bounded hot-key table before the main job using record, byte, or cost thresholds, then broadcast that small table to choose the salted path. Version the list by data date and give it a threshold, count limit, and fallback. This adds a planning stage and operational state, so stable SLA gains must justify the complexity.

If the slow operation is groupBy, do you still replicate a dimension?

No. For a mergeable aggregation, salt hot rows, compute partial aggregates by (key, salt), then merge those partials by key. That distributes one hot key's input across tasks, while the second stage processes a small number of partials. State whether the aggregate is safely mergeable. Globally ordered or non-mergeable state cannot use the technique without a different algorithm.

How do you choose 32 salt buckets?

Divide hot-partition bytes by target task input for a lower bound, then account for available cores, small-side replication, scheduler overhead, and output-file constraints. If 720 GiB should fall to about 32 GiB per task, the theoretical lower bound is roughly 23, so 32 is a reasonable experiment in this scenario. Compare several candidates on maximum task input, stage time, and executor-hours, and keep limited headroom for hotspot growth.

Can speculative execution solve this straggler?

A duplicate of a deterministic skewed task still reads the same 720 GiB partition, usually repeating expensive work on two executors. Speculation is more useful for an intermittently slow node or transient jitter. Check whether the same task stays slow after retrying elsewhere. If its input remains the outlier, split the data work instead of duplicating it.