Spark Internals & Performance

Driver and executors, shuffle, AQE, join strategies, skew and salting, plus a ten-step method for debugging a slow Spark job.

Same format as Parts 1–3, but this one opens with a theory primer because Spark questions in senior interviews almost always start with "explain how X works internally" before moving to "now here's a slow job, debug it." Get the mental model solid first — the Q&A after it will click faster if the theory underneath it is automatic.

THEORY PRIMER

Cluster architecture

A Spark application has one Driver (runs your main program, holds the SparkSession, builds the DAG, schedules work) and many Executors (JVM processes on worker nodes that actually run tasks and hold cached data / shuffle files). The driver doesn't do the heavy data processing — it plans and coordinates; executors do the work. On Databricks, the driver and executors run on the job/all-purpose cluster's driver node and worker nodes respectively. A Task is the smallest unit of work — one task per partition per stage. A Stage is a set of tasks that can run without a shuffle boundary between them. A Job (Spark job, not a Databricks Job — different concept, same word) is triggered by an action (.collect(), .write(), .count()) and consists of one or more stages.

Lazy evaluation and the DAG

Transformations (select, filter, join, groupBy) are lazy — calling them just builds up a logical plan (a DAG of operations), nothing executes. Only an action (write, collect, count, show) triggers actual execution. This matters practically: Spark can optimize the whole chain of transformations before running anything (predicate pushdown, column pruning, join reordering), and it's why a slow "cell" in a notebook might actually be paying for several upstream lazy transformations that only materialize at that action.

Catalyst optimizer and Tungsten

Catalyst is the query optimizer — it takes your DataFrame/SQL logical plan, applies rule-based optimizations (predicate/filter pushdown, constant folding, column pruning) and cost-based optimizations (join reordering based on table statistics), then generates a physical plan (choosing, e.g., which join strategy to use). Tungsten is the execution engine underneath — off-heap binary memory format (avoids JVM object overhead and GC pressure), whole-stage code generation (compiles a chain of operators into a single JVM bytecode function instead of interpreting each operator with virtual-function-call overhead), and cache-aware algorithms. Together these are why DataFrame/SQL API is generally faster than hand-written RDD code doing the same logical thing — you get the optimizer and the codegen for free.

RDD vs DataFrame vs Dataset vs SQL

RDD is the low-level distributed-collection abstraction — no schema, no Catalyst optimization, full control but you're on your own for performance. DataFrame is a distributed table with a schema, goes through Catalyst/Tungsten — this is the default I reach for. Dataset adds compile-time type safety (Scala/Java only, not Python) on top of DataFrame — still Catalyst-optimized. Spark SQL is just another way to express the same logical plan as DataFrame code — they compile to the identical physical plan, so "SQL vs DataFrame API" is a readability/team-convention choice, not a performance one.

Partitions and parallelism

Data is split into partitions, each processed by one task. Partition count determines parallelism — too few partitions means idle cores (a 64-core cluster processing 8 partitions is running at 8/64 utilization); too many means scheduling overhead and small-task overhead dominates actual work. Default shuffle partition count is 200 (spark.sql.shuffle.partitions) — a notoriously bad one-size-fits-all default that's frequently wrong for both small and very large datasets, and is exactly what Adaptive Query Execution (below) was built to fix automatically.

Shuffle

A shuffle is a full redistribution of data across the cluster — every executor writes shuffle files to disk, and every executor in the next stage reads the files relevant to its partitions over the network. It's the most expensive operation in Spark: disk I/O to write shuffle files, network I/O to fetch them, and a stage boundary (the next stage can't start until the shuffle write is complete). Anything that needs to bring rows with the same key together onto the same partition — groupBy, join (non-broadcast), distinct, repartition, orderBy — triggers one.

Narrow vs Wide Transformations

map
filter
    ↓
narrow

groupBy
join
distinct
    ↓
wide
shuffle

Narrow: each output partition depends on exactly one (or a known, fixed subset of) input partition(s) — map, filter, union, mapPartitions. No data movement across executors needed; these can execute within a single stage, pipelined together with no shuffle boundary.

Wide: an output partition can depend on data from any input partition — groupBy, join (unless broadcast), distinct, repartition, orderBy, reduceByKey. Requires a shuffle, which means a new stage.

This distinction is the foundation of Spark's stage planning: Catalyst chains consecutive narrow transformations into one stage (pipelined, no intermediate materialization), and inserts a stage boundary at every wide transformation.

Adaptive Query Execution (AQE)

Enabled by default in modern Spark/Databricks Runtime. AQE re-optimizes the query plan at runtime, using actual statistics from completed stages rather than only pre-execution estimates — it can: coalesce small shuffle partitions into fewer, right-sized ones (fixing the "200 is wrong for this data size" problem automatically), convert a sort-merge join to a broadcast join if runtime stats show one side is actually small, and split skewed partitions (see skew handling below). This is a big deal for the "why is my job slow" conversation — a lot of classic manual-tuning advice (hand-picking shuffle.partitions) matters less than it used to, though it's still worth knowing how to override it when AQE's heuristics get it wrong.

Caching / Persistence

.cache() / .persist(StorageLevel) materializes a DataFrame in memory (and/or disk, depending on storage level) so repeated actions on it don't recompute the whole lineage from source. Only worth it when the same DataFrame is used multiple times downstream (e.g. branching into two different aggregations) — caching something used exactly once just adds overhead. Storage levels trade memory vs recomputation vs serialization cost (MEMORY_ONLY, MEMORY_AND_DISK, MEMORY_AND_DISK_SER); on Databricks, Delta Cache (disk-based, automatic for Delta reads on supported instance types) often makes manual .cache() less necessary for reading the same Delta table repeatedly.

Memory management

Executor memory splits into execution memory (shuffles, joins, sorts, aggregations — transient working memory) and storage memory (cached data) — unified in modern Spark, meaning they can borrow from each other under pressure, with execution generally able to evict cached storage data if it needs the space. When execution memory isn't enough for an operation (e.g. a huge shuffle/sort/join), Spark spills intermediate data to disk — this is safe (won't crash) but slow (disk I/O where you wanted memory speed), and heavy spill in the Spark UI is one of the clearest "this needs more memory or better partitioning" signals.

Q&A

Narrow vs Wide Transformations

Q: Why does the narrow/wide distinction matter for performance, beyond just "wide is slower"?

A: It determines stage boundaries and fault-recovery granularity. Narrow transformations pipeline within a stage — no materialization between them, so filter then map then filter again costs almost nothing extra beyond the base scan. Wide transformations force a stage boundary: all upstream tasks must complete and write shuffle output before any downstream task can start, which also means a straggler task in the upstream stage delays the entire downstream stage from starting. Minimizing unnecessary wide transformations (e.g. avoiding a repartition you don't actually need, or reordering filters to happen before a join/groupBy so less data is shuffled) is one of the highest-leverage tuning moves.

Joins

Q: You have a 2 TB fact table and a 10 MB dimension table. How do you join them?

A: Broadcast the dimension table — broadcast(dim_df) or let AQE do it automatically (it will, given accurate/runtime-derived size stats, if the table is under spark.sql.autoBroadcastJoinThreshold, default 10MB, though AQE's runtime detection is more reliable than the static threshold for this). The dimension gets serialized and sent to every executor once; each executor then does the join locally against its partition of the fact table — no shuffle of the 2TB table at all. This is the single most impactful join optimization when there's a large-small size asymmetry, and it's the answer I'd want to give in under two seconds in an interview because it's exactly this common in real pipelines (fact/dimension is the canonical case).

Q: What is a sort-merge join and when does Spark use it?

A: The default strategy for large-large joins where neither side fits comfortably for broadcast. Both sides are shuffled (partitioned by join key) and sorted within each partition, then merged — like a merge-sort merge step, walking both sorted sides together matching keys. Cost: a full shuffle of both tables plus the sort. It's the general-purpose fallback that works at any scale but is the most expensive option when a cheaper strategy (broadcast) would've worked.

Q: What is a shuffle hash join and how does it differ from sort-merge?

A: Both sides are shuffled by join key (like sort-merge), but instead of sorting, Spark builds an in-memory hash table on one side (the smaller of the two, post-shuffle) and probes it with the other side — no sort step. Faster than sort-merge when the build side fits in memory per-partition, but riskier: if the build side is bigger than expected (skew, bad stats), you get OOM instead of the more gradual degrade-and-spill behavior sort-merge has. Spark's planner prefers sort-merge by default for large-large joins precisely because it's more memory-robust; shuffle hash is chosen in narrower circumstances (one side notably smaller, but too big to broadcast).

Q: What are the limitations of broadcast joins?

A: The broadcast side must fit in the memory of every executor simultaneously (it's shipped to all of them) — too large and you OOM the executors rather than get a slow-but-working query. It's also a full materialize-then-ship operation, so even a "small" table that's expensive to compute (deep upstream transformation chain) adds that cost up front. And broadcast only helps the join step itself — if the broadcast side needs to be built from a wide/expensive upstream query, you've just moved the cost rather than eliminated it. I also watch for accidentally broadcasting something large — e.g. autoBroadcastJoinThreshold estimating table size from stale stats and misjudging a table as small; I've had to explicitly disable broadcast (spark.sql.autoBroadcastJoinThreshold = -1 or df.hint("merge")) for a specific problem join where Spark's auto-broadcast guess was wrong and kept OOMing.

Q: What's a skewed join and how do you fix it?

A: When join keys are unevenly distributed (a handful of keys have vastly more rows than others — classic example: a "null" or "unknown" customer_id catching millions of unmatched rows, or a popular product ID in an orders table), the partition(s) holding those keys become enormous while others are tiny — one or a few tasks run far longer than the rest, and the whole stage waits on them. Fixes: AQE skew join optimization (enabled by default in modern Spark — it detects oversized partitions post-shuffle and automatically splits them into sub-partitions, joining each sub-partition against a duplicated copy of the matching other-side partition); manual salting (append a random suffix to the skewed key on both sides within a matching range, artificially spreading one hot key across multiple partitions) for cases AQE doesn't fully resolve; or isolating the skewed keys — filter them out, broadcast-join just those separately (since it's typically a small number of distinct hot keys even if huge in row count), union the results back.

Q: Walk me through the salting technique concretely.

A: Add a column salt = rand() * N (say N=10) to the large/skewed side, replicate the small side N times with an explicit salt value 0..N-1 attached to each replica (a cross-join-like explode, cheap because the small side is small), then join on (original_key, salt) instead of just original_key. This spreads what was one massive partition (all rows for the hot key) into N smaller partitions, at the cost of N-way replication of the small side — a reasonable trade when the size asymmetry is large.

The Systematic Debugging Sequence — "Why is this Spark job slow?"

This is the capstone question for this section. Walk it in order, explaining what you're looking for at each step and what it would tell you — don't just list the ten items, narrate the decision tree.

1. Spark UI
2. Query plan
3. Shuffle
4. Data skew
5. Partition count
6. Join strategy
7. Serialization
8. File sizes
9. Storage access
10. Cluster configuration

1. Spark UI — start here, always. Jobs → Stages tab: which stage is actually slow (don't guess — the UI tells you precisely). Look at task duration distribution within that stage: if most tasks finish fast and a few run long, that's a straggler/skew signal, not a "the whole job is just slow" problem, and the fix is completely different.

2. Query plan — .explain(True) or the SQL tab's visualized DAG. Confirms what physical operations are actually happening — is it doing the join strategy I expect? Is a filter getting pushed down before a join/scan, or is it (wrongly) happening after, forcing more data through an expensive step than necessary? Are there unexpected extra shuffles from something like an accidental repartition or a groupBy before it's needed?

3. Shuffle — check shuffle read/write size per stage in the UI. A huge shuffle relative to source data size is a red flag — either the join/groupBy is touching more data than it should (missing a filter pushdown, joining before filtering instead of after), or it's the wrong join strategy for the size asymmetry (should've been broadcast).

4. Data skew — task duration histogram in the stage detail view; a long tail of a handful of tasks taking 10-100x the median is the signature. Cross-reference with the join/groupBy key's actual value distribution if you can — is there a dominant/null key?

5. Partition count — too few partitions (check df.rdd.getNumPartitions() or the stage's task count) means idle cores; too many means task-scheduling overhead dominating actual work (each task has fixed overhead — if tasks finish in milliseconds, you have too many). Compare partition count to cluster core count — you generally want partition count to be a small multiple of total cores, not equal to a fixed default that ignores actual cluster size or data volume.

6. Join strategy — confirmed from the query plan (step 2): is it broadcasting when it should, or doing an expensive sort-merge on a case that should've been a broadcast? Is AQE enabled and actually kicking in (check Spark UI for "AQEShuffleRead" nodes indicating it did rewrite the plan)?

7. Serialization — default Java serialization is slow and produces large shuffle payloads; Kryo (spark.serializer = org.apache.spark.serializer.KryoSerializer) is significantly faster and more compact for shuffle/cache data — relevant mainly for RDD-heavy code or custom objects; DataFrame/Dataset operations mostly bypass this concern via Tungsten's binary format, so this step matters more when there's RDD code, UDFs operating on complex objects, or heavy .cache() usage in the pipeline.

8. File sizes — the small-file problem from the Delta section: thousands of tiny files means thousands of tiny tasks with fixed per-task overhead dominating, plus expensive file-listing/metadata overhead on read. Conversely, a handful of huge files can limit parallelism (can't split below a file in some formats/configs) and cause a few tasks to do disproportionate work. Check via DESCRIBE DETAIL on the Delta table (numFiles, average file size) — this is often the actual root cause when steps 1-7 all look individually reasonable but the job is still slow.

9. Storage access — is the read pattern hitting storage efficiently? Predicate/partition pruning actually happening (confirm in the query plan — are irrelevant partitions being skipped, or is it scanning the whole table)? For cloud storage specifically: throttling (watch for 503s/retries in the driver/executor logs — ADLS/S3 request-rate limits can silently degrade a job that looks compute-bound but is actually waiting on storage retries), and whether Delta's data-skipping stats are actually useful for the filters being applied (irrelevant if the filtered column isn't in the first 32 stats-tracked columns or isn't Z-Ordered for a high-cardinality filter).

10. Cluster configuration — last, because it's the "throw resources at it" lever, and I want to have ruled out an actual inefficiency in the plan before reaching for it. Right executor sizing (too few cores/executors relative to partition count = underutilization; too many small executors = more shuffle service overhead and network chatter), memory sizing relative to actual data volume (spill in the UI is direct evidence of undersized executor memory for the operation), and whether autoscaling is fighting the workload (a job with a big shuffle mid-pipeline might not benefit from scaling up, since the shuffle stage boundary limits how much added parallelism actually helps once shuffle I/O — not just CPU — is the bottleneck).

Closing framing for this answer: "I go UI-first, not config-first — nine times out of ten the fix is in the query plan or the data (skew, tiny files, wrong join strategy), and throwing more cluster at a plan-level problem just makes it fail slightly less slowly and cost more. Cluster sizing is the last lever, not the first one."

Serialization (standalone questions)

Q: When does serialization actually become a bottleneck in a Spark job?

A: Shuffle (every shuffled row gets serialized for network transfer and deserialized on the receiving side), caching with a serialized storage level, and any UDF/RDD operation on non-primitive JVM objects. It's rarely the top issue for straightforward DataFrame SQL/API pipelines (Tungsten's binary format mostly sidesteps generic object serialization), but becomes real when there's custom Scala/Java object logic, heavy Python UDFs (different problem — see below), or RDD-based code inherited from an older codebase.

Q: What's the performance concern specifically with Python/Pandas UDFs vs native Spark SQL functions?

A: A plain Python UDF forces row-by-row serialization across the JVM↔Python process boundary (py4j), which is slow and defeats Tungsten/whole-stage codegen entirely for that step. Pandas UDFs (vectorized, Arrow-based) batch rows into Arrow record batches for the JVM↔Python handoff, which is dramatically faster than row-at-a-time — if a UDF is unavoidable, Pandas UDF over plain UDF, every time. But the first choice is always: can this be expressed with native Spark SQL functions (when, expr, built-in string/date/array functions) instead of a UDF at all — native functions get full Catalyst optimization and codegen, UDFs are always at least partially opaque to the optimizer.

File Sizes / Small Files

Q: How do small files hurt performance beyond "more tasks"?

A: Fixed per-task overhead (scheduling, JVM task setup) dominates when actual work per task is tiny — a 10MB file processed in a task with \~50-100ms of fixed overhead spends a large fraction of wall-clock time on overhead, not I/O or compute. On top of that, listing/opening many small files against cloud storage (ADLS/S3) has its own per-file latency, so the driver's file-listing phase before the job even starts can become significant, and for Delta specifically, tracking many small files bloats the transaction log's per-commit metadata. Fix: OPTIMIZE (from Part 2), and controlling write-time file sizing (spark.databricks.delta.optimizeWrite.enabled, target file size config) so the problem doesn't accumulate in the first place.

Storage Access

Q: What does "storage access" as a bottleneck actually look like in practice, and how do you distinguish it from a compute bottleneck?

A: In the Spark UI, tasks that are slow but show low CPU time relative to wall-clock duration — the task is waiting, not computing. Common causes: cloud storage throttling (retryable 503s consuming wall time in retry/backoff), a poorly partitioned/pruned read forcing a full scan instead of touching only relevant files, or reading many small files (previous point) where I/O overhead per file dominates. I'd check executor logs for retry/throttling messages and compare the ratio of task CPU time to task duration in the UI's task metrics — a big gap there points at storage/network wait, not compute.

Data Skew (deeper questions beyond the join case)

Q: Does skew only happen in joins, or can groupBy/aggregation suffer from it too?

A: Same underlying cause, different operation — groupBy(key).agg(...) shuffles by key just like a join does, so a handful of keys with disproportionate row counts create the same oversized-partition problem. AQE's skew handling applies here too for the shuffle read side. Salting works similarly for a skewed aggregation, though it requires a two-phase aggregate (aggregate per salted-sub-key first, then aggregate those partial results by the true key) rather than the join-side replication trick.

Q: How do you detect skew before it becomes a production incident, rather than debugging it after the job is already slow?

A: Track key-distribution stats (e.g. groupBy(key).count() sorted descending, as a periodic data-quality/profiling check) on any column used as a join or groupBy key in a production pipeline, especially ones fed by external/upstream sources where the distribution can silently shift over time — a key that was evenly distributed at pipeline design time isn't guaranteed to stay that way as the business/data changes.

Quick-fire recap