Apache Spark Internals & Troubleshooting

Architecture: driver, executors, cluster manager

Every Spark application has exactly one driver and a set of executors. The driver runs your program's main(): it builds the logical plan, talks to the cluster manager to acquire resources, splits work into tasks, and schedules those tasks onto executors. Executors are JVM processes on worker nodes that run tasks on partitions of data and hold cached blocks. The cluster manager (on Databricks, an internal manager - you never configure standalone/YARN/Kubernetes yourself) allocates the executor processes.

On Databricks classic compute, the mapping is concrete: the driver runs on the driver node of the cluster, one executor per worker node by default, and notebook cells execute on the driver until an action forces distributed work. This is why a df.collect() or a giant toPandas() kills clusters - it pulls all data back into the driver JVM's memory. With serverless compute (GA for notebooks, jobs, and pipelines) the same logical roles exist but Databricks manages the sizing.

Lazy evaluation, the DAG, and jobs → stages → tasks

Transformations (select, filter, join, groupBy) are lazy: they only extend a logical plan. Nothing executes until an action (count, write, collect, show) triggers a job. Spark then compiles the plan into a DAG and cuts it into stages at every point where data must move between executors - that is, at every wide transformation (joins, groupBy aggregations, repartition, window functions over non-colocated keys, distinct). Narrow transformations (filter, map, column projection) pipeline together inside a single stage because each output partition depends on one input partition.

Each stage runs as a set of tasks, one task per partition, and a task is the unit a single executor core processes. So the hierarchy is: one action → one job (sometimes more) → stages separated by shuffle boundaries → tasks equal to the partition count of that stage. Hold this model in your head and the Spark UI stops being noise: a slow job is a slow stage, a slow stage is either too many/too few tasks or a handful of pathological tasks.

Catalyst: from your code to a physical plan

Catalyst is the query compiler behind both DataFrames and Spark SQL - they hit the same optimizer, which is why "DataFrame vs SQL performance" is a non-question. Four phases:

1. Parse - SQL text or DataFrame calls become an unresolved logical plan (column names not yet bound).
2. Analyze - names and types are resolved against the catalog (on Databricks, Unity Catalog).
3. Optimize - rule-based rewrites: predicate pushdown, column pruning, constant folding, join reordering. This is why filtering "late" in your code usually doesn't matter - Catalyst pushes the filter down anyway.
4. Physical planning - Catalyst generates candidate physical plans (e.g., broadcast hash join vs sort-merge join), costs them using statistics, and picks one.

Whole-stage codegen then collapses the operators of a stage into a single generated Java function, eliminating virtual calls and per-row interpretation. In df.explain() output, operators marked with * are codegen'd. On Databricks, Photon (the native vectorized C++ engine) takes over much of this work where supported - worth name-dropping in an interview, but know that the Catalyst phases are identical either way.

Adaptive Query Execution (AQE)

Static plans guess; AQE corrects the guesses at runtime using real shuffle statistics from completed stages. Enabled by default on modern Databricks Runtime, it does three big things:

• Coalesces shuffle partitions. Instead of you tuning spark.sql.shuffle.partitions per query, AQE merges small post-shuffle partitions into sensibly sized ones. The static setting becomes an upper bound, not a target.
• Switches join strategies. If a join side turns out smaller than the broadcast threshold after filters run, AQE converts a planned sort-merge join into a broadcast hash join mid-flight.
• Handles skewed joins. AQE detects shuffle partitions far larger than the median and splits them into subpartitions, replicating the matching side, so one giant key doesn't pin a single task for an hour.

AQE does not fix everything: it can't help skewed aggregations as thoroughly as skewed joins, it can't broadcast something that's still too big, and it operates only at shuffle boundaries. Know its limits as well as its features.

Shuffle: why it dominates cost

A shuffle redistributes rows across executors by key. Mechanically: each map task partitions its output by hash of the key, sorts/serializes it, and writes shuffle files to local disk; reduce tasks then fetch their slice from every map task over the network, deserialize, and often sort again. Disk I/O, network I/O, serialization, and memory pressure - all in one operation. That's why a job's cost profile is usually "the shuffles, plus rounding error."

Practical consequences: minimize shuffle count (don't repartition casually, pre-aggregate before joins where valid), minimize shuffle width (prune columns early so less data crosses the wire), and prefer broadcast joins where one side is genuinely small.

Partition sizing rules of thumb

• Target roughly 128-200 MB per partition of in-flight data. Smaller means scheduler overhead and tiny files; much larger means spill and slow stragglers.
• spark.sql.shuffle.partitions defaults to 200 - almost never right for big workloads when AQE is off. With AQE on (the Databricks default), leave it generous and let coalescing shrink it; Databricks' auto-optimized shuffle can manage it for you entirely.
• Task count should comfortably exceed total executor cores (2-3x is a common heuristic) so the scheduler can balance, but not by orders of magnitude.
• repartition(n) triggers a full shuffle and can increase or decrease partitions evenly; coalesce(n) only merges existing partitions without a shuffle - cheap, but can produce unevenly sized partitions and reduce parallelism for everything before the write.

# repartition vs coalesce before a write
# Full shuffle: even partitions, costs a stage
df.repartition(64, "fiscal_period").write.format("delta").save(path)

# No shuffle: just merges partitions - fine for shrinking
# output file count, but upstream stages keep old parallelism
df.coalesce(8).write.format("delta").save(path)

Join strategies

Spark picks a physical join from statistics and hints. The three you must know cold:

from pyspark.sql.functions import broadcast

# Fact-to-dimension: force broadcast of the small side
gl_detail = spark.read.table("silver.gl_transactions")     # billions of rows
coa = spark.read.table("silver.chart_of_accounts")         # a few MB

result = gl_detail.join(broadcast(coa), "account_id", "left")
# SQL equivalent: SELECT /*+ BROADCAST(coa) */ ...

Data skew: detection and fixes

Skew is uneven key distribution: a few partitions carry most of the rows, so a few tasks run forever while the cluster idles. Real financial data is skew by construction - intercompany accounts, default cost centers, and null business keys concentrate enormous row counts on single join keys.

Detect it in the Stages tab: open the task-level summary metrics and compare min / median / max for duration and shuffle read size. Healthy stages have max close to the 75th percentile; a max 50x the median is skew. A stage stuck at "199/200 tasks complete" for twenty minutes is the classic smell.

Fix it, in order of preference:

1. Let AQE handle it - skew-join optimization splits oversized partitions automatically. Confirm in the plan that it actually fired.
2. Broadcast the other side - skew is irrelevant if there's no shuffle on the skewed side.
3. Filter or separate hot keys - nulls and sentinel keys often shouldn't join at all; union a hot-key path with the normal path.
4. Salting - manually spread a hot key across N synthetic subkeys, replicating the small side N times.

from pyspark.sql import functions as F

SALT_N = 16
# Big, skewed side: scatter each key across 16 salted keys
big_salted = big.withColumn(
    "salted_key",
    F.concat_ws("_", "join_key", (F.rand() * SALT_N).cast("int"))
)
# Small side: replicate every row once per salt value
salts = spark.range(SALT_N).withColumnRenamed("id", "salt")
small_salted = small.crossJoin(salts).withColumn(
    "salted_key", F.concat_ws("_", "join_key", "salt")
)
joined = big_salted.join(small_salted, "salted_key")

Spill: what it is and how to spot it

Spill happens when an operator's working set (sort buffers, aggregation hash maps, join hash tables) exceeds the execution memory available to the task, so Spark serializes blocks to disk and reads them back later. The job still succeeds - that's the point of spill - but you pay serialization plus disk I/O both ways, often multiplying stage time several times over.

In the Spark UI, look at Spill (Memory) and Spill (Disk) in stage/task metrics. Memory spill is the in-memory size of spilled data; disk spill is its serialized size on disk (smaller, since serialized). Any non-zero spill on a hot stage is a tuning signal. Causes and fixes: partitions too large (raise partition count / let AQE coalesce less aggressively), skew (see above), too many cores per executor sharing too little memory (fewer, fatter tasks), or an oversized aggregation that should be staged. Throwing bigger instance types at spill works but is the expensive fix - check partition sizing first.

Reading the Spark UI: a practical walkthrough

A repeatable triage sequence for a slow Databricks job:

1. Jobs tab. Which job in the run is eating the wall-clock time? Note its duration and how many stages it has. Many short jobs in a loop is a different problem (driver-side orchestration) than one long job.
2. SQL / DataFrame tab. For the slow query, open the graph. This is the highest-signal view: per-operator row counts, the join strategies actually chosen, every Exchange (shuffle), and whether AQE rewrote anything. A scan reading 2 billion rows into a filter that keeps 2 million tells you pushdown or file layout failed.
3. Stages tab. For the dominant stage, read the summary metrics table - min/25th/median/75th/max for duration, shuffle read, shuffle write, spill. Skew, spill, and tiny-task overhead are all visible here in one screen.
4. Task metrics. Shuffle Read/Write sizes tell you how heavy the boundary was; GC time as a large fraction of task time signals memory pressure (often co-occurring with spill); scheduler delay signals too many tiny tasks.
5. Executors tab. Dead executors, uneven task distribution, storage memory usage from caching. On Databricks, pair this with cluster metrics for CPU/memory saturation.

cache() / persist(): semantics and pitfalls

df.cache() is persist(StorageLevel.MEMORY_AND_DISK) for DataFrames: partitions are materialized the first time an action computes them, stored in executor memory and overflowing to disk, and reused by later actions. Key semantics:

• Caching is lazy - nothing is stored until an action runs. df.cache(); df.count() is the common materialization idiom.
• It helps only when the same DataFrame is reused by multiple actions (iterative logic, a branch point feeding several outputs). A cache used once is pure overhead.
• Cached blocks consume the same unified memory pool execution needs - over-caching causes eviction (recompute later) and increases spill pressure elsewhere.
• Release it with df.unpersist() when done; check the Storage tab to see what's actually resident.

This module is the foundation for everything downstream in the DE track: Delta Lake's OPTIMIZE, clustering, and file-size management are largely about making the scan and shuffle story above cheaper before Spark even starts.