Top Apache Spark interview questions, with answers

Apache Spark is a distributed engine for large-scale data processing. You describe a computation over a dataset, and Spark splits both the data and the work across a cluster of machines, coordinating the whole thing from a single process called the driver.

Three properties define it:

• Distributed execution. A dataset is divided into partitions, and each partition is processed independently, in parallel, on whichever machine has capacity.
• Lazy evaluation. Operations build a plan rather than running immediately, which lets the engine optimise the whole computation before any of it starts.
• Fault tolerance through lineage. Spark records how each partition was derived, so a lost one is rebuilt rather than the job being restarted.

It ships APIs for Scala, Java, Python and R, plus libraries for SQL, structured streaming, machine learning and graph processing, all on the same execution engine.

The framing that lands well in an interview: Spark is a compute engine, not a storage system and not a table format. It reads from HDFS, S3, JDBC or Kafka, and writes back out. Confusing Spark with Hudi, Iceberg or Delta is a common mistake, and being crisp about the boundary is a good signal.

Because one of two things has become true: the data no longer fits the memory, CPU or disk of one machine, or it fits but takes unacceptably long to process serially.

Spark’s answer to both is partitioning. A 1 TB dataset might become 8,000 partitions of roughly 128 MB, and if the cluster has 200 cores then 200 of them are processed at a time.

1 TB input
     |
     v  split into partitions
+------+------+------+     +--------+
|  P1  |  P2  |  P3  | ... |  P8000 |
+------+------+------+     +--------+
     |                          |
     v   processed in parallel  v
        across 200 cores
     |
     v
   result

The part candidates usually miss is the cost. Distribution is not free: there is serialization, network transfer during shuffles, scheduling overhead per task, and the coordination cost of the driver. For a dataset that fits comfortably on one machine, a single-process tool such as pandas, DuckDB or Polars will routinely beat a Spark cluster, because Spark’s overhead is not amortised.

A good answer names the crossover point rather than claiming Spark is always faster. Spark earns its keep when the work genuinely exceeds one machine, or when you need fault tolerance across a long-running job.

The structural difference is where intermediate results live and how much of the computation the engine can see at once.

MapReduce runs one map phase and one reduce phase, and writes the intermediate result to disk, usually replicated, between them. A multi-step algorithm becomes a chain of MapReduce jobs, each re-reading the previous one’s output from disk.

Spark keeps intermediate data in memory where it can, and expresses an entire chain of operations as one plan before running any of it. That gives it two advantages MapReduce structurally cannot have:

The iterative case is where the gap is widest. Ten passes of an algorithm in MapReduce means ten round trips through disk; in Spark the dataset can be cached once and reused.

The honest framing for an interview: the advantage is in-memory intermediate data plus whole-plan optimisation, not a fixed speed multiplier. The often-quoted “100x” comes from a specific iterative benchmark and does not describe a typical batch job. Quote a number only if you measured it on your own data, and otherwise describe the mechanism.

Transformations do not compute anything when you call them. Each one records what to do and returns a new dataset describing the result. Nothing executes until an action asks for a value.

The RDD programming guide states it directly:

All transformations in Spark are lazy, in that they do not compute their results right away. Instead, they just remember the transformations applied to some base dataset.

Why it matters is the important half. Because Spark holds the whole chain before running it, Catalyst can rewrite it: push a filter down into the scan so fewer rows are read, drop columns nobody selected, fuse adjacent operations into one generated loop, and choose a join strategy knowing what comes next. None of that is possible for an engine that executes each call as it arrives.

flowchart TB
  T["transformations called:<br/>read, filter, withColumn, groupBy"]
  T --> P[("plan recorded<br/><b>0 jobs, 0 tasks</b>")]
  P -.->|"nothing has run"| W["cluster is idle"]
  A["an action is called:<br/>count()"] --> R["<b>1 job</b>: Catalyst optimises<br/>the whole plan, then runs it"]
  P --> R

There are two consequences to be ready for:

• A dataset is recomputed every time an action needs it, unless you persist it. The guide is explicit: “each transformed RDD may be recomputed each time you run an action on it.”
• Errors surface at the action, not at the transformation. A typo in a column name in a filter may only raise when you call count(), which is why analysis errors seem to appear far from their cause.

A transformation creates a new dataset from an existing one and runs nothing. An action returns a value to the driver or writes to storage, and that is what submits a job.

Transformations divide further, and the distinction drives everything about performance:

• Narrow transformations, where each input partition feeds at most one output partition, so the work pipelines inside a single task: map, filter, mapPartitions.
• Wide transformations, where an output partition draws from many input partitions, so data must move across the network: groupByKey, join, distinct, repartition.

The count of actions tells you how many jobs you will run, and the count of wide transformations tells you how many stages each job will have.

One trap worth naming: show() and take(n) are actions, so calling them during development runs the whole chain up to that point. A notebook with ten show() calls has run ten jobs, and if nothing is cached, much of the work is repeated each time.

Count the jobs. Chaining transformations submits none; the action submits one.

sc = spark.sparkContext
rdd = sc.parallelize(range(4), 2)

doubled = rdd.map(lambda x: x * 2).filter(lambda x: x > 2)
print(len(sc.statusTracker().getJobIdsForGroup() or []))   # 0

doubled.count()
print(len(sc.statusTracker().getJobIdsForGroup() or []))   # 1

Two transformations, zero jobs. One action, one job.

A second demonstration is more visual: put a side effect in a transformation and watch when it happens.

acc = sc.accumulator(0)
counted = sc.parallelize(range(100), 4).map(lambda x: (acc.add(1), x)[1])

print(acc.value)      # 0, the map has not run
counted.count()
print(acc.value)      # 100, it ran when the action asked

The same trick exposes the other half of laziness. Call count() a second time and the accumulator reads 200, for the same 100 rows, because the uncached dataset was recomputed from scratch. Add .cache() and it stays at 100.

In an interview, the accumulator version is the stronger answer, because it demonstrates laziness and recomputation in one go.

A partition is the unit of parallelism: a slice of the dataset that one task processes on one core. A stage runs exactly one task per partition, so the partition count is the maximum parallelism available to that stage.

flowchart TB
  subgraph S["one stage"]
    P1["partition 0"] --> T1["task 0"]
    P2["partition 1"] --> T2["task 1"]
    P3["partition 2"] --> T3["task 2"]
    P4["partition 3"] --> T4["task 3"]
  end
  T1 --> E["executors run<br/>as many as they have slots"]
  T2 --> E
  T3 --> E
  T4 --> E

Three numbers follow from this, and almost every tuning question is one of them:

• How many partitions. Too few and cores sit idle; too many and per-task scheduling overhead dominates. A useful target is partitions in the low hundreds of megabytes, with several tasks per core so stragglers average out.
• How evenly the data is spread. One partition holding most of the rows is skew, and because a task cannot be split, that one task sets the stage duration.
• Where they come from. On read, from file splitting; after a shuffle, from spark.sql.shuffle.partitions or adaptive coalescing.

Check the number rather than assume it:

print(df.rdd.getNumPartitions())

SparkSession is the entry point for the DataFrame and SQL APIs, and it is what modern code creates. SparkContext is the older, lower-level entry point for RDDs and for cluster coordination.

from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("trips-ingest")
    .config("spark.sql.shuffle.partitions", "400")
    .getOrCreate())

sc = spark.sparkContext          # the SparkContext inside it

A SparkSession wraps exactly one SparkContext, and one application has one SparkContext. getOrCreate() returns the existing one rather than making a second, which is why it is the idiomatic call.

Two details interviewers probe:

• Creating the session is what starts negotiating for executors. Before it, nothing has been requested from the cluster manager.
• newSession() gives an isolated SQL session, with its own temporary views and SQL configs, sharing the same SparkContext and therefore the same executors. That is how a multi-tenant server such as the Thrift server isolates users.

Scala, Java, Python and R, with a community-maintained C# binding outside the project. For DataFrame and SQL work the choice barely affects performance, because operations in any language compile to the same logical plan and execute in the JVM.

It matters in two places:

• RDD code, where the logic itself is a closure in your language. A Python RDD map ships every row to a Python worker process and back.
• Python UDFs, which sit inside a plan that is otherwise all JVM. Rows must be serialized out to Python and results serialized back, and because Catalyst cannot see inside the function, it cannot push filters through it or fuse it into generated code.

# stays in the JVM, fully optimisable
df.filter(F.col("fare_amount") > 50)

# leaves the JVM per batch, breaks codegen across this point
@F.pandas_udf("double")
def adjust(v: pd.Series) -> pd.Series:
    return v * 1.1

The ordering to remember, best first: built-in functions and SQL, then pandas (vectorised) UDFs which amortise the crossing over an Arrow batch, then plain Python UDFs which pay it per row.

Scala and Java additionally get the typed Dataset API. Python does not, since it has no compile-time types to check.

Three nested units of work, each created by a different component.

flowchart TB
  A["action: write()"] --> J["1 job"]
  J --> S0["stage 0<br/>read, filter, partial aggregate"]
  S0 -->|"shuffle"| S1["stage 1<br/>final aggregate, write"]
  S0 --> T0["8 tasks<br/>one per input partition"]
  S1 --> T1["200 tasks<br/>one per shuffle partition"]

Read it as a hierarchy: a job has one or more stages, and a stage has one task per partition. The arithmetic is worth internalising, because it is how you predict what the UI will show:

• Jobs equals the number of actions you called.
• Stages per job equals the number of shuffles, plus one.
• Tasks per stage equals that stage’s partition count, unrelated to how many executors you have.

A nuance that catches people: a single write can produce more than one job, because some writers run extra jobs for committing or for computing statistics. “One action, one job” is the right model, with that exception noted.

A directed acyclic graph of dependencies between datasets. Directed because every edge points from an input to an output; acyclic because nothing can depend on itself, which is what makes the graph safe to walk.

Spark builds it as you chain transformations, and DAGScheduler uses it for two jobs:

1. Cutting stages. It walks the graph backwards from the action and starts a new stage at every shuffle boundary, because everything narrow can pipeline into the stage it is already in.
2. Recovery. Because each node records how it was derived, a lost partition can be rebuilt by replaying just the path that produced it.

flowchart TB
  A["read trips"] --> B["filter"]
  C["read cities"] --> D["filter"]
  B --> J["join"]
  D --> J
  J --> K["groupBy city"]
  J --> L["groupBy rider"]
  K --> M["write by city"]
  L --> N["write by rider"]

The DAG is the reason Spark is not restricted to a map phase and a reduce phase: any shape of dependency graph is expressible, so a job can branch, rejoin and have many stages.

You can print it for an RDD with toDebugString(), and for a DataFrame the physical plan from explain("formatted") is the equivalent. In the UI, the DAG visualisation on the job page shows the same thing graphically, with each box a stage.

The driver is the process that holds the SparkSession and makes every decision. In order, it:

1. Creates the SparkSession, and with it the SparkContext, SparkEnv and the RPC endpoints executors will call back on.
2. Requests executors from the cluster manager through SchedulerBackend.
3. Turns your code into a logical plan, has Catalyst optimise it, and selects a physical plan.
4. Cuts the plan into stages in DAGScheduler, at every ShuffleDependency.
5. Places individual tasks on executors via TaskSchedulerImpl and TaskSetManager, preferring the best available locality.
6. Tracks every attempt, retrying failures up to spark.task.maxFailures, default 4.
7. Maintains BlockManagerMaster, the index of every cached and shuffle block in the cluster.
8. Receives results, bounded in total by spark.driver.maxResultSize, default 1g.
9. Serves the UI on port 4040 and writes the event log the history server replays.

What it does not do is process your data. It is a coordinator.

Two consequences follow, and both are common interview questions:

• spark.driver.memory defaults to 1g, which is a coordinator’s budget. Anything that pulls data back, collect() or toPandas(), needs more, or needs not to do that.
• The driver is the one process with no redundancy. The plan, the scheduler state and the block index exist nowhere else, so if the driver dies the application dies, whereas losing an executor costs only recomputable work.

An executor is a JVM launched for one specific application that runs tasks and stores blocks. The Spark source describes it in a line: a “Spark executor, backed by a threadpool to run tasks”.

Its lifecycle:

1. Registers. CoarseGrainedExecutorBackend sends RegisterExecutor and waits for RegisteredExecutor before accepting work. Until that round trip finishes, a granted container is doing nothing.
2. Accepts LaunchTask messages and runs each in a pool thread. Concurrency is spark.executor.cores, which defaults to 1 on YARN and to every core on the machine in standalone mode.
3. Reads input, runs the generated code, and for a shuffle map stage writes shuffle output to local disk.
4. Reports back with StatusUpdate per task and a heartbeat every spark.executor.heartbeatInterval, default 10s. Miss enough and the driver declares it lost.
5. Stores and serves blocks through its BlockManager, which is what makes a cached DataFrame reusable and lets the next stage fetch shuffle data from its peers.

The contract is deliberately narrow: an executor never makes a scheduling decision, never sees the query plan, and never coordinates with another executor about what to run. It receives tasks, runs them, and serves bytes.

That narrowness is what makes executors disposable. Lose one and its tasks are retried elsewhere and its cached partitions are recomputed from lineage. Executors are also per-application: two applications never share one, which is why memory is reserved per application rather than pooled.

A cluster manager is a separate service whose only job is to decide who gets containers. At Spark 4.x the supported options are Standalone, YARN and Kubernetes.

Mesos support was removed in the 4.x line. At v4.2.0 the resource-managers/ directory in the source tree contains only kubernetes and yarn, so the three rows above are the whole list.

The division worth internalising is that the cluster manager never schedules your tasks. It hands out containers; executors register back with the driver; and from that point the driver drives everything. That single fact explains a lot of otherwise puzzling behaviour:

• Why an under-provisioned driver throttles a large cluster.
• Why YARN’s own scheduler settings change what you are given but never how your tasks are distributed.
• Why Kubernetes needs no Spark daemon at all: the driver talks to the API server directly.

Kubernetes is the clearest illustration of the split, because the component granting resources and the component scheduling work are visibly different things.

In the general sense yes: one process coordinates and many do the work. But Spark’s own vocabulary is worth getting exactly right, because “master” is overloaded and people conflate three different things with it.

Spark has also moved away from the older terminology in its own scripts. At v4.2.0 there is sbin/start-workers.sh and conf/workers.template, while start-slaves.sh and slaves.template are gone, and the standalone documentation speaks of “a master and workers”.

Where the analogy breaks is the interesting part, and it is a strong thing to volunteer. In a classic master-worker system the master assigns work. In Spark the cluster manager does not assign your tasks: it grants containers, and the driver schedules into them. The coordinator role is split, resources from one component and work from another.

The script that launches an application against any supported cluster manager through one uniform interface, so the same jar runs on YARN, Kubernetes or standalone with only --master changing.

spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 4g \
  --executor-cores 4 \
  --conf spark.sql.shuffle.partitions=400 \
  examples/jars/spark-examples.jar 1000

The options fall into three groups: what to run (--class, the jar), where to run it (--master, --deploy-mode), and how much to ask for (--executor-memory, --executor-cores, --num-executors).

The mistake to know: every Spark-facing option must come before the application jar. Anything after the jar is passed to your application’s main as an argument instead, so a misplaced --conf is silently ignored by Spark and handed to your code.

Configuration precedence, highest first:

1. Values set on SparkConf in your code.
2. Flags passed to spark-submit.
3. spark-defaults.conf.

--properties-file names an alternative defaults file, and --load-spark-defaults makes Spark read spark-defaults.conf as well as that file. --verbose prints where each value came from, which settles most “why is my config not applying” questions.

Where the driver runs. That is the whole difference, and everything else follows from it.

flowchart TB
  subgraph C["client mode"]
    C1["submitting machine<br/><b>driver here</b>"] --> C2["cluster manager"]
    C2 --> C3["executors in the cluster"]
    C1 <-->|"all scheduling traffic"| C3
  end
  subgraph K["cluster mode"]
    K1["submitting machine<br/>exits after submit"] --> K2["cluster manager"]
    K2 --> K3["<b>driver in the cluster</b>"]
    K3 <--> K4["executors in the cluster"]
  end

Two practical consequences:

• Client mode from a laptop over a VPN is slow and fragile, because every scheduling message and every collected result crosses that link. Cluster mode keeps it local.
• In client mode the driver JVM already exists by the time your code runs. So spark.driver.extraJavaOptions set in SparkConf arrives too late to affect the JVM. Use --driver-java-options or a properties file instead. This catches people trying to add a logging config or a heap dump flag.

There are two separate settings, and confusing them is a classic slip.

spark.default.parallelism governs RDD operations. Its documented default depends on the operation: for a distributed shuffle such as reduceByKey it is the largest number of partitions in the parent RDD, and for an operation with no parent, such as parallelize, it follows the cluster manager, which on local mode is the number of cores.

spark.sql.shuffle.partitions governs DataFrame and SQL shuffles, and it defaults to 200 regardless of your data or your cluster.

print(spark.conf.get("spark.sql.shuffle.partitions"))   # 200

So if you are tuning a DataFrame job and you change spark.default.parallelism, nothing happens, which is a frequent source of confusion. The other thing to know is that input read parallelism is governed by neither: it comes from file splitting via spark.sql.files.maxPartitionBytes.

It is the number of partitions produced by a DataFrame or SQL shuffle, and it defaults to 200.

The trouble is that 200 is a constant, while the right value depends on your data volume and your core count. Both directions hurt:

• Too few for the data. 200 partitions over 1 TB means 5 GB per task, which will spill to disk and may fail outright.
• Too many for the data. 200 partitions over 100 MB means 500 KB per task, so scheduling and task startup dominate and 200 near-empty files get written.
• Mismatched to cores. With 32 cores, 200 tasks is a sensible several-waves-per-core; with 2,000 cores, most sit idle.

A reasonable target is partitions in the low hundreds of megabytes, and a total that is a small multiple of your core count so stragglers average out.

Adaptive query execution changes the calculus. With spark.sql.adaptive.enabled and spark.sql.adaptive.coalescePartitions.enabled both true by default, Spark writes 200 shuffle partitions and then merges the small ones at runtime toward spark.sql.adaptive.advisoryPartitionSizeInBytes, default 64MB.

spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

So in modern Spark the better habit is to set the advisory size and leave the partition count high, letting coalescing do the arithmetic, rather than guessing a single number for every stage in the query.

Spark 3.3.0. The configuration file became log4j2.properties and the JVM flag became -Dlog4j.configurationFile=.

It matters because the syntax changed completely, and most logging guides on the internet predate it. Log4j 1 used a flat log4j. prefix:

# Log4j 1, ignored by Spark 3.3.0 and later
log4j.rootLogger=INFO, console
log4j.logger.org.apache.spark=WARN

Log4j 2 uses named components:

# Log4j 2
rootLogger.level = info
rootLogger.appenderRef.stdout.ref = console
appender.console.type = Console
appender.console.name = console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{HH:mm:ss} %-5level %logger{1} - %msg%n
logger.spark.name = org.apache.spark
logger.spark.level = warn

At v4.2.0 the only template Spark ships is conf/log4j2.properties.template; the Log4j 1 template is gone. So a config written in the old syntax is not an error you will see, it is a file that is silently ignored, which is the worst kind of failure.

Ship it to both sides, since driver and executor answer different questions:

spark-submit \
  --files /etc/spark/conf/log4j2.properties \
  --conf "spark.driver.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties" \
  --conf "spark.executor.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties" \
  app.jar

A Resilient Distributed Dataset: an immutable, partitioned collection of records that can be operated on in parallel, carrying enough information to rebuild any lost partition.

Unpack the name, because each word is a design decision:

• Resilient. Spark records how each partition was derived, not just the data. Lose one and it is recomputed from its parents rather than the job failing.
• Distributed. The collection is split into partitions spread across executors.
• Dataset. A collection of records, with no schema and no interpretation of their contents.

It is also immutable. Every transformation produces a new RDD rather than modifying the old one, which is what makes lineage a reliable recovery record: a parent cannot have changed underneath you.

rdd = spark.sparkContext.parallelize([("sf", 24.5), ("nyc", 61.75)], 2)
totals = rdd.reduceByKey(lambda a, b: a + b)

The trade against DataFrames is visibility. Because an RDD holds opaque objects, Spark cannot see inside your closures, so there is no schema, no column pruning, no predicate pushdown and no Catalyst. You get control and lose optimisation, which is why the DataFrame API is the default for structured data.

From RDD.scala, every RDD is characterised by five things, three required and two optional:

1. A list of partitions. getPartitions, which defines the parallelism.
2. A function for computing each partition. compute, which is the actual work.
3. A list of dependencies on other RDDs. getDependencies, which is the lineage.
4. Optionally, a partitioner for key-value RDDs, such as HashPartitioner.
5. Optionally, preferred locations for each partition, such as the HDFS block locations.

The first three make the RDD work; the last two make it fast, and they are the ones worth talking about:

• The partitioner is what lets Spark skip a shuffle. If two RDDs share a partitioner on the join key, matching keys are already co-located, so the join is narrow.
• Preferred locations are what data locality is built on. TaskSchedulerImpl tries to place a task where its input already is, which is why locality levels exist.

rdd = spark.sparkContext.parallelize(range(10), 4)
print(rdd.getNumPartitions())                    # 1
print(rdd.partitioner)                           # 4, None for a non-pair RDD

Naming all five, and then saying which two are optional and why they matter, is a much stronger answer than listing them.

Through lineage: the chain of transformations from a source to the dataset in hand. Because Spark recorded how to build every partition, not just the partition itself, a lost one is rebuilt on a surviving executor.

lineage = (spark.sparkContext.parallelize(range(10), 2)
    .map(lambda x: x + 1)
    .filter(lambda x: x % 2 == 0))
print(lineage.toDebugString().decode())

Two things make this cheaper than it sounds:

• Recovery is per partition, not per job. Losing one executor costs the partitions it held, which are recomputed while everything else stands.
• Only the path matters. Spark replays the branch that produced the lost partition, not the whole graph.

flowchart TB
  S["source"] --> A["map"] --> B["filter"] --> C["partition 3 lost"]
  C -.->|"recompute just this path"| D["partition 3 rebuilt<br/>on another executor"]

The limits are worth volunteering:

• Shuffle output is the exception. If the shuffle files a completed stage wrote die with their executor, DAGScheduler resubmits that entire stage, because the next stage has nothing to fetch. That is a stage rerun rather than a task retry, and it is why a job can appear to go backwards.
• Long lineages get expensive to replay, which is what checkpointing exists to fix: it writes the data to reliable storage and truncates the chain.
• A non-deterministic transformation breaks the guarantee, because a recomputed partition may not match the lost one.

In a narrow dependency each parent partition feeds at most one child partition, so the work can pipeline inside a single task with no data movement. In a wide dependency a child partition draws from many parents, which requires a shuffle.

flowchart TB
  subgraph N["narrow: map, filter"]
    P0["P0"] --> C0["C0"]
    P1["P1"] --> C1["C1"]
    P2["P2"] --> C2["C2"]
  end
  subgraph W["wide: groupByKey, join"]
    Q0["P0"] --> D0["C0"]
    Q0 --> D1["C1"]
    Q1["P1"] --> D0
    Q1 --> D1
    Q2["P2"] --> D0
    Q2 --> D1
  end

The recovery difference is the part people miss. A lost narrow partition needs one parent partition recomputed. A lost partition after a wide dependency may need data from every parent partition, which is far more expensive, and is another reason shuffles are costly beyond their immediate network cost.

coalesce is the interesting edge case: it is narrow, because several parent partitions feed one child without a shuffle, which is exactly why it can only reduce the count and may leave partitions uneven.

At every ShuffleDependency in the lineage, and nowhere else.

DAGScheduler walks the graph backwards from the action. Each time it crosses a ShuffleDependency it ends the current stage and starts a new one; everything narrow is absorbed into the stage it is already in.

flowchart TB
  subgraph S0["stage 0"]
    R["read"] --> F["filter"] --> W["withColumn"] --> PA["partial aggregate"]
  end
  PA -->|"ShuffleDependency"| SH[("shuffle files<br/>on local disk")]
  subgraph S1["stage 1"]
    SH --> FA["final aggregate"] --> O["write"]
  end

The practical consequences:

• A long chain of narrow operations costs no extra stages. Twenty chained filter and withColumn calls still run in one stage, fused by whole-stage codegen into a single loop.
• The stage count is the shuffle count plus one, per job. Counting Exchange nodes in a physical plan gives you the same number.
• Every stage but the last is a ShuffleMapStage, whose purpose is to produce shuffle output. The last is a ResultStage, which computes what the action asked for.

Stages also enforce a barrier: stage 1 cannot start until stage 0 has finished writing all its output, because a reduce task needs its slice from every map task. That is why one straggler in stage 0 delays the entire next stage, and why skew is so expensive.

reduceByKey combines values on the map side before the shuffle; groupByKey shuffles every value and combines afterwards.

# prefer this: combines locally first
pairs.reduceByKey(lambda a, b: a + b)

# shuffles every value across the network
pairs.groupByKey().mapValues(sum)

flowchart TB
  subgraph R["reduceByKey"]
    A1["('sf',1) ('sf',1) ('sf',1)"] --> A2["local combine<br/>('sf',3)"] --> A3["shuffle<br/>1 record"]
  end
  subgraph G["groupByKey"]
    B1["('sf',1) ('sf',1) ('sf',1)"] --> B3["shuffle<br/>3 records"] --> B4["combine after"]
  end

With a hot key, the difference is the number of records crossing the network: one per key per partition for reduceByKey, versus every record for groupByKey. groupByKey also has a correctness-adjacent risk: all values for one key must fit in memory on the reduce side, so a hot key can cause an OutOfMemoryError that reduceByKey would not.

The subtlety that makes a strong answer: a map-side combine disqualifies Spark’s two faster shuffle writers, because they cannot combine. So reduceByKey shuffles less data through a more expensive writer. It is still almost always the right choice, because network volume dominates, but the trade-off is real and worth naming.

Use groupByKey only when you genuinely need every value, for example to collect them into a list, and even then consider aggregateByKey which combines into a buffer as it goes.

map applies your function once per element. mapPartitions applies it once per partition, handing you an iterator over that partition’s elements and expecting an iterator back.

# once per row: a connection per row, which is fatal
rdd.map(lambda r: lookup(open_conn(), r))

# once per partition: one connection amortised over thousands of rows
def enrich(rows):
    conn = open_conn()
    for r in rows:
        yield lookup(conn, r)
    conn.close()

rdd.mapPartitions(enrich)

Use mapPartitions when there is per-partition setup worth amortising: a database or HTTP connection, loading a model, or compiling a regex.

Two cautions:

• Do not materialise the partition. list(rows) inside the function loads the whole partition into memory and defeats the streaming design. Use a generator, as above.
• Return an iterator, not a single value. Returning a list works but holds everything in memory.

mapPartitionsWithIndex additionally gives you the partition number, which is the standard trick for inspecting how data is distributed:

print(rdd.mapPartitionsWithIndex(
    lambda i, rows: [(i, sum(1 for _ in rows))]).collect())

That prints rows per partition, which is how you confirm skew on an RDD.

map returns exactly one output per input. flatMap returns an iterable per input and flattens the results, so it can return zero, one or many.

lines = sc.parallelize(["a b c", "d e"])

lines.map(lambda s: s.split()).collect()
# [['a','b','c'], ['d','e']]        two elements, each a list

lines.flatMap(lambda s: s.split()).collect()
# ['a','b','c','d','e']             five elements

The “zero” case is the one worth knowing, because it makes flatMap a combined map and filter:

# parse, and silently drop anything unparseable, in one pass
rdd.flatMap(lambda s: [parse(s)] if valid(s) else [])

Both are narrow transformations, so neither causes a shuffle, and both pipeline into the surrounding stage. The DataFrame equivalent of flatMap for a column is explode, which turns an array column into one row per element.

repartition does a full shuffle to reach any target count and balances the result. coalesce merges existing partitions without a full shuffle, so it can only reduce the count and may leave partitions uneven.

df.coalesce(10).write.parquet(path)      # cheap, before a write
df.repartition(400).groupBy("k").count() # balanced, before heavy work
df.repartition("city_id")                # hash-partitioned by a column

The trap is what coalesce does upstream, and it is a favourite interview question. Because coalesce is narrow, it does not create a stage boundary, so the reduced parallelism propagates backwards into the stage feeding it. coalesce(1) before a heavy transformation can make that entire transformation run in a single task.

flowchart TB
  subgraph C["coalesce(2): merge, no shuffle"]
    A1["P0 10 MB"] --> M1["P0 + P1<br/>30 MB"]
    A2["P1 20 MB"] --> M1
    A3["P2 5 MB"] --> M2["P2 + P3<br/>45 MB"]
    A4["P3 40 MB"] --> M2
  end
  subgraph R["repartition(2): shuffle, balanced"]
    B1["P0 10 MB"] --> SH[("shuffle")]
    B2["P1 20 MB"] --> SH
    B3["P2 5 MB"] --> SH
    B4["P3 40 MB"] --> SH
    SH --> N1["37 MB"]
    SH --> N2["38 MB"]
  end

So if you need one output file, repartition(1) pays for a shuffle but keeps the upstream stage parallel. coalesce(1) is only safe when the work before it is trivial.

cache() is exactly persist() with the default storage level. persist(level) lets you choose the level.

df.cache()                                    # default level
df.persist(StorageLevel.MEMORY_AND_DISK_SER)  # explicit

The default differs by API, which is the part that surprises people:

So an RDD cache that exceeds memory silently recomputes, which looks like the cache doing nothing at all. A DataFrame cache spills instead, which is slower than memory but still avoids recomputation. If an RDD cache seems to have no effect, that asymmetry is usually why.

Both are lazy: they mark the dataset and populate it on the next action. The exception is SQL’s CACHE TABLE, which is eager by default.

Release it explicitly when you are done, because cached blocks hold storage memory for the life of the application and squeeze execution memory from every later stage:

df.unpersist()
spark.catalog.clearCache()       # everything

The levels combine three choices: memory or disk, serialized or not, and replicated or not.

How to choose:

• Serialized forms are far more compact, so more fits in memory, at the cost of CPU to deserialize on every read. Worth it when memory is the constraint.
• _2 replicas put a copy on a second node, so losing an executor does not force recomputation. Worth it when recomputation is very expensive; otherwise it doubles the memory cost.
• DISK_ONLY is right when recomputing is more expensive than reading back, which is the case after a long or costly lineage.
• OFF_HEAP avoids garbage collection pressure but needs spark.memory.offHeap.enabled, which defaults to false, and a size.

Read the storage tab of the UI to check what actually happened. A dataset showing well below 100% in memory is being partly recomputed or read from disk on every use, which usually means the cache is not paying for itself.

A broadcast variable is a read-only value sent to each executor once and cached there, rather than serialized into every task.

The guide describes them as keeping “a read-only variable cached on each machine rather than shipping a copy of it with tasks”.

sc = spark.sparkContext
cities = sc.broadcast({"sf": "San Francisco", "nyc": "New York"})

resolved = sc.parallelize(["sf", "nyc", "sf"], 2).map(lambda c: cities.value[c])
print(resolved.collect())   # ['San Francisco', 'New York', 'San Francisco']

Without the broadcast, that dictionary is captured by the closure and serialized with every task. With 10,000 tasks and a 10 MB lookup table, that is 100 GB of serialization instead of one copy per executor.

flowchart TB
  subgraph W["captured variable"]
    D1["driver"] -->|"copy per task<br/>x 10,000"| T1["tasks"]
  end
  subgraph B["broadcast"]
    D2["driver"] -->|"one copy per executor<br/>x 20"| E2["executors"] --> T2["all tasks read locally"]
  end

Practical notes:

• Read-only. Mutating .value on an executor changes only that copy and nothing comes back.
• Call .unpersist() when finished, or .destroy() to release it permanently, if it is large and no longer needed.
• This is the same machinery behind a broadcast hash join, which is why spark.sql.autoBroadcastJoinThreshold exists and why broadcasting something large pressures the driver, which must assemble it first.

Accumulators are write-only-from-tasks counters that the driver can read. Tasks add to them; only the driver should read the value.

errors = sc.accumulator(0)

def parse(line):
    try:
        return json.loads(line)
    except ValueError:
        errors.add(1)
        return None

parsed = rdd.map(parse).filter(lambda x: x is not None)
parsed.count()
print(errors.value)

The catch is the guarantee, and it is the whole question. The guide is explicit:

For accumulator updates performed inside actions only, Spark guarantees that each task’s update to the accumulator will only be applied once, i.e. restarted tasks will not update the value. In transformations, users should be aware of that each task’s update may be applied more than once if tasks or job stages are re-executed.

This is not theoretical, and it follows directly from lazy evaluation:

acc = sc.accumulator(0)
counted = sc.parallelize(range(100), 4).map(lambda x: (acc.add(1), x)[1])

counted.count(); print(acc.value)   # 100
counted.count(); print(acc.value)   # 200, for the same 100 rows

The uncached dataset is recomputed for the second action, so the accumulator increments again. Add .cache() and it stays at 100. A task retry has the same effect.

So use accumulators for diagnostics you can tolerate being approximate, and never as the source of a number the output depends on. If you need an exact count, compute it as data with an aggregation.

A partitioner is the function mapping a key to a partition number. Spark ships two:

• HashPartitioner, which uses hash(key) % numPartitions. The default for reduceByKey and friends.
• RangePartitioner, which samples the data to find range boundaries so that partitions are roughly equal and ordered. Used by sortByKey and by a global orderBy.

It helps because a known partitioner lets Spark skip a shuffle. If two RDDs share the same partitioner on the join key, matching keys are already co-located, so the join becomes a narrow dependency.

part = 100
a = rdd_a.partitionBy(part).cache()      # note the cache
b = rdd_b.partitionBy(part).cache()
joined = a.join(b)                       # no shuffle

The cache is not optional. Without persisting, the partitioning is recomputed each time the RDD is used, so you pay the shuffle you were trying to avoid.

You can inspect and set it:

print(pairs.partitioner)                      # None until you set one
tagged = pairs.partitionBy(HashPartitioner(100))

The DataFrame equivalents are repartition(col) for the same effect within a job, and bucketing for the effect to persist across jobs, recorded in the metastore.

It prints the lineage of an RDD as an indented tree: each level is an RDD, with its type, its id and its partition count.

lin = (sc.parallelize(range(100), 4)
    .map(lambda x: (x % 5, x))
    .reduceByKey(lambda a, b: a + b))
print(lin.toDebugString().decode())

(4) PythonRDD[3] at RDD at PythonRDD.scala:53 []
 |  MapPartitionsRDD[2] at mapPartitions at ... []
 |  ShuffledRDD[1] at partitionBy at ... []
 +-(4) PairwiseRDD[0] at reduceByKey at ... []
    |  ParallelCollectionRDD ... []

How to read it:

• The number in parentheses is the partition count at that level.
• Indentation changes, marked by +-, are shuffle boundaries. Count them and you have the stage count minus one.
• Read bottom up: the source is at the bottom, the RDD you called it on is at the top.

It is the fastest way to answer “how many stages will this take” for RDD code, and to spot an accidental shuffle. For DataFrames the equivalent is df.explain("formatted"), where each Exchange is the same signal.

When you need something the structured APIs do not expose. The honest list is short:

• Custom partitioning logic beyond hash or range, for example routing by a business rule.
• Per-partition resource handling via mapPartitions, although the DataFrame API has mapInPandas for much of this now.
• Genuinely unstructured data with no schema to impose: raw text, binary records, custom parsing where a row shape only emerges after processing.
• Low-level control over lineage and checkpointing, in iterative algorithms.

For everything with a schema, the DataFrame API wins, and the reason is structural rather than a matter of taste: Catalyst can see a DataFrame plan and cannot see inside an RDD closure. That means no column pruning, no predicate pushdown, no join strategy selection and no whole-stage codegen for RDD code. Tungsten’s compact binary rows also avoid materialising JVM objects, which RDD code cannot avoid.

A good answer also notes that RDDs did not go away. DataFrames are built on them: df.rdd gives you the underlying RDD, and the scheduler works entirely in RDD terms. So understanding RDDs is still how you understand what a DataFrame job actually does.

collect() brings every row to the driver. take(n) returns the first n rows and tries to read as little as possible to get them.

rows = df.collect()     # everything, bounded only by driver memory
head = df.take(5)       # 5 rows, usually from one partition
df.show(5)              # take plus formatting

take(n) starts by scanning one partition. If that yields fewer than n rows it scans more, increasing the number it tries each round. So on a large dataset it usually touches a tiny fraction of the data.

collect() is bounded by spark.driver.maxResultSize, default 1g, and by the driver heap, spark.driver.memory, default 1g. Exceeding the first raises an error naming the config; exceeding the second is an OutOfMemoryError.

This is the classic way to kill a driver, and the answer an interviewer wants is what to do instead:

• To inspect, use show(), take() or limit().
• To count, use count(), which aggregates rather than transferring rows.
• To export, use write so executors write in parallel and only a status returns.
• To hand data to Python, toPandas() has the same problem as collect(); prefer writing out or using mapInPandas to keep the work distributed.

Both aggregate an RDD to a single value using a binary function. The difference is the zero value.

sc.parallelize([1, 2, 3, 4]).reduce(lambda a, b: a + b)      # 10
sc.parallelize([1, 2, 3, 4]).fold(0, lambda a, b: a + b)     # 10
sc.parallelize([]).reduce(lambda a, b: a + b)                # raises
sc.parallelize([]).fold(0, lambda a, b: a + b)               # 0

So fold is defined on an empty RDD and reduce is not.

The trap is that the zero value must be a true identity, because Spark applies it once per partition as well as when combining partitions. With four partitions and a zero of 1 for addition, you would add 1 five times, not once.

aggregate generalises both: it takes a zero, a function to fold a partition into an accumulator, and a function to combine accumulators. That is what you need when the accumulator type differs from the element type, such as computing a mean:

total, n = sc.parallelize([1, 2, 3, 4]).aggregate(
    (0, 0),
    lambda acc, x: (acc[0] + x, acc[1] + 1),      # fold within a partition
    lambda a, b: (a[0] + b[0], a[1] + b[1]))      # combine partitions
print(total / n)

Note also that all three are actions, returning to the driver, whereas reduceByKey is a transformation returning a distributed RDD.

Because only pair RDDs can be partitioned by key, and key-partitioning is what enables the optimisations that matter.

Being a pair RDD unlocks:

• partitionBy, and therefore shuffle-free joins between co-partitioned RDDs.
• Map-side combines: reduceByKey, aggregateByKey, combineByKey.
• Key-wise operations: groupByKey, join, cogroup, lookup, sortByKey.

pairs = rdd.map(lambda r: (r.city_id, r.fare_amount))   # now a pair RDD
pairs.reduceByKey(lambda a, b: a + b)

In Scala these arrive by implicit conversion on RDD[(K, V)] through PairRDDFunctions, which is why they are not on the RDD class itself. In Python any RDD whose elements are 2-tuples qualifies.

The distinction matters in an interview because it explains why the API is shaped this way: partitioning is defined on keys, so anything that benefits from co-location has to be expressed as a pair. The DataFrame equivalent is that these optimisations attach to columns you group or join on, which is why repartition(col) and bucketing exist.

It is retried, and if retries are exhausted the stage and then the job fail.

The sequence:

1. The task fails and reports back to the driver.
2. TaskSetManager reschedules it, preferring a different executor if the failure looked executor-specific.
3. Retries continue up to spark.task.maxFailures, default 4, counted per task across the stage.
4. On the fourth failure the stage fails, and the job fails with it.

Two related mechanisms are worth naming:

• Blacklisting, now called excludeOnFailure. If one executor or node keeps failing tasks, Spark can stop scheduling onto it, which prevents a single bad machine from burning through the retry budget of every task.
• Speculation. spark.speculation, default false, duplicates tasks that are running much slower than their peers and takes whichever finishes. It helps with a slow node and hurts with skew, because the duplicate is equally slow.

The distinct case is losing shuffle output rather than a task:

flowchart TB
  T1["task attempt 1 fails"] --> T2["attempt 2, preferably<br/>on another executor"]
  T2 --> T3["attempt 3"]
  T3 --> T4["attempt 4"]
  T4 --> F["spark.task.maxFailures reached:<br/><b>stage fails, job fails</b>"]
  T2 -.->|"same executor keeps failing"| X["excludeOnFailure:<br/>stop scheduling there"]
  T2 -.->|"slow, not failing"| SP["speculation may<br/>duplicate the task"]

That is a stage rerun, and it is why a job can appear to go backwards in the UI. An external shuffle service avoids it by serving shuffle blocks independently of the executor that wrote them.

A distributed collection of rows organised into named columns, with a schema Spark knows at plan time. That last part is the whole point.

df = spark.read.parquet("s3a://lakehouse-prod/warehouse/trips")
df.printSchema()

Because Spark knows the schema and the operations, a DataFrame plan goes through Catalyst, which can:

• Prune columns nobody selected, so they are never read from disk.
• Push predicates into the scan, so rows are filtered at the source.
• Choose a join strategy based on estimated sizes.
• Fuse operators into a single generated loop via whole-stage codegen.
• Store rows in Tungsten’s binary format, avoiding JVM objects entirely.

An RDD of the same data gets none of that, because its records are opaque objects and its closures are opaque functions.

Two clarifications interviewers like:

• A DataFrame in Scala is Dataset[Row]. It is not a separate type; Row is the untyped record.
• It is still immutable and lazy. withColumn returns a new DataFrame, and nothing runs until an action.

A Dataset is a typed distributed collection that still runs through Spark SQL’s optimised engine. It gives you compile-time type safety and your own case classes instead of generic rows.

case class Trip(tripId: String, fareAmount: Double, cityId: String)

val trips: Dataset[Trip] = spark.read.parquet(path).as[Trip]
trips.filter(_.fareAmount > 50).map(_.cityId)   // typed, checked at compile time

In Scala, DataFrame is simply Dataset[Row], so they are the same machinery with different static types.

Python has no typed Dataset API, and the reason is worth stating: the API exists to give compile-time checking, and Python has no compile step to check in. In PySpark, DataFrame is what you get.

The cost of typed Datasets is real too: a lambda like _.fareAmount > 50 is opaque to Catalyst in the same way an RDD closure is, so it cannot be pushed down. Typed filter and map trade optimisation for type safety, which is why even in Scala, column expressions are often preferred for hot paths.

How to choose, in one line each:

• DataFrame or SQL for almost everything. It is the fastest path because Catalyst can see it all.
• Dataset in Scala or Java when compile-time safety on a domain type is worth losing pushdown through typed lambdas.
• RDD only for custom partitioning, per-partition resource handling, or data with no schema.

The evolution is worth a sentence: RDDs came first, DataFrames added a schema and the optimiser, and Datasets added types back on top of the optimiser. They are layered, not alternatives, and df.rdd drops down a level whenever you need it.

The strong closing point: hand-written RDD code loses to DataFrame code on structured data not because the author wrote it badly, but because the engine is allowed to rewrite the DataFrame version and is not allowed to touch the RDD one.

Catalyst is Spark SQL’s query optimiser: the component that turns a query you wrote into a physical plan the engine runs.

It is a rule-based tree transformation framework. Plans are trees, rules are functions from tree to tree, and the optimiser applies batches of rules until the tree stops changing. That design is why Spark SQL is extensible: a new optimisation is a new rule, and you can add your own through session extensions.

flowchart TB
  A["SQL or DataFrame"] --> B["unresolved<br/>logical plan"]
  B -->|"catalog"| C["analysed<br/>logical plan"]
  C -->|"rules"| D["optimised<br/>logical plan"]
  D -->|"strategies + cost"| E["physical plan"]
  E -->|"codegen"| F["JVM bytecode"]

What Catalyst gives you in practice, in rough order of value:

• Column pruning, so unread columns are never touched.
• Predicate pushdown, so filtering happens at the source.
• Join strategy selection, so a small side gets broadcast.
• Constant folding and expression simplification.
• Whole-stage code generation, producing tight loops rather than operator-by-operator iteration.

The idea underneath all of it is lazy evaluation. Catalyst can only optimise a plan it can see in full, which is exactly what laziness provides.

Five, and naming them in order with what each does is a strong answer.

1. Parse. SQL text or DataFrame calls become an unresolved logical plan: a tree with relation and column names that have not been checked against anything.
2. Analyse. The analyzer resolves table names, column names, data types and function calls against the catalog. This is where AnalysisException comes from, and why a typo in a column name surfaces here rather than at parse time.
3. Logical optimisation. Rule batches rewrite the plan: predicate pushdown, column pruning, constant folding, boolean simplification, filter and projection reordering, subquery flattening. The output is still logical, meaning it says what, not how.
4. Physical planning. Strategies turn each logical operator into one or more physical ones, for example a logical join into a broadcast hash join, shuffle hash join or sort-merge join. Costs choose between candidates.
5. Code generation. Whole-stage codegen fuses the operators in a stage into one generated Java method, compiled with Janino, so intermediate rows never become objects.

You can see each stage:

df.explain("extended")   # parsed, analysed, optimised, physical, in that order

The distinction to be crisp about is logical versus physical. Logical optimisation is algebraic and knows nothing about the cluster; physical planning is where partitioning, shuffles and join implementations are decided.

Moving a filter as close to the data as possible, so that fewer rows are ever read or carried up the plan.

There are two levels, and distinguishing them is a good signal:

• Within the plan. Catalyst moves a Filter below a Project or through a join, so downstream operators see fewer rows.
• Into the data source. For Parquet or ORC, the predicate is handed to the reader, which skips entire row groups using footer statistics. For JDBC, it becomes a WHERE clause in the SQL sent to the database.

You can confirm it in the plan:

(spark.read.parquet(path)
   .filter(F.col("fare_amount") > 50)
   .select("city_id")
   .explain("formatted"))

FileScan parquet [fare_amount#19,city_id#21]
  DataFilters: [isnotnull(fare_amount#19), (fare_amount#19 > 50.0)]
  PushedFilters: [IsNotNull(fare_amount), GreaterThan(fare_amount,50.0)]
  ReadSchema: struct<fare_amount:double,city_id:string>

PushedFilters is the line that matters: those predicates reached the file format. spark.sql.parquet.filterPushdown defaults to true.

What blocks it is the practically useful half:

• Wrapping the column in a function. to_date(ts) = '2026-09-17' cannot push down; ts >= '...' AND ts < '...' can.
• A UDF in the predicate, because Catalyst cannot see inside it.
• Filtering after a non-deterministic operation, which would change semantics.

Reading only the columns the query actually needs, and dropping the rest as early as possible in the plan.

On a columnar format such as Parquet or ORC, unread columns are never touched on disk at all, so a table with 200 columns queried for 3 reads roughly 1.5% of the bytes. On a row format such as JSON or CSV the whole row must still be read and parsed, so pruning helps less.

Confirm it by reading ReadSchema in the plan:

ReadSchema: struct<fare_amount:double,city_id:string>

Two columns, not the whole table.

What defeats it:

• select("*"), explicitly asking for everything.
• df.rdd or a typed Dataset lambda, which forces Spark to materialise the whole row because it cannot see which fields you touch.
• Caching before selecting. df.cache() caches the columns the plan needed at that point, so caching a wide DataFrame and then selecting three columns stores all of them.

The practical habit: select the columns you need immediately after reading, not at the end of the pipeline. It costs nothing and it is the difference between reading 1.5% and 100% of a wide table.

An optimisation where Catalyst fuses all the operators in a stage into a single generated Java method, rather than running each operator as a separate iterator that passes rows to the next.

Without it, a scan-filter-project chain is three objects calling next() on each other, with a virtual call and an intermediate row per step. With it, Spark emits roughly the loop you would write by hand:

// conceptually, for filter + project over a scan
while (scan.hasNext()) {
  InternalRow row = scan.next();
  double fare = row.getDouble(0);
  if (!row.isNullAt(0) && fare > 50.0) {
    out.write(row.getUTF8String(1));
  }
}

You can see it in the plan. Operators marked with * and a number share one generated stage:

*(1) HashAggregate(keys=[city_id#21], ...)
+- Exchange hashpartitioning(city_id#21, 200)
   +- *(2) HashAggregate(keys=[city_id#21], ...)
      +- *(2) Project [city_id#21]
         +- *(2) Filter (fare_amount#19 > 50.0)
            +- *(2) FileScan parquet

Everything marked *(2) was fused into one method.

What breaks the fusion is the useful part to know: a Python UDF, because rows must leave the JVM; some complex expressions; and plans that exceed the generated-method size limit, at which point Spark falls back to the interpreted path. Losing codegen is a large part of why Python UDFs cost what they do.

Start with the mode that matches the question:

df.explain("formatted")   # the readable one: tree plus per-node details
df.explain("extended")    # parsed, analysed, optimised, physical
df.explain("cost")        # with statistics, where available
df.explain(True)          # same as extended

Then read the physical plan bottom up, because that is execution order: scans at the bottom, then filters and projections, then exchanges, then the final operator.

What to look for, in the order that usually matters:

The last row matters for reading plans in modern Spark. Before execution the plan is wrapped in AdaptiveSparkPlan with isFinalPlan=false, meaning what you see is the planned shape. To see what actually ran, look at the SQL tab of the UI after the query finishes, where the final plan shows the strategies AQE settled on.

select projects a set of columns, replacing the projection entirely. withColumn adds or replaces one column, keeping everything else.

df.select("trip_id", (F.col("fare_amount") * 1.1).alias("fare_with_fee"))
df.withColumn("fare_with_fee", F.col("fare_amount") * 1.1)

For one or two columns the difference is style. For many it is performance, and this is a real production issue:

# 50 chained calls: a 50-level deep plan, slow to analyse and optimise
for i in range(50):
    df = df.withColumn(f"c{i}", F.col("x") + i)

# one projection: a flat plan
df = df.select("*", *[(F.col("x") + i).alias(f"c{i}") for i in range(50)])

Each withColumn adds a Project node. Catalyst will usually collapse them, but the analysis and optimisation passes still walk the deep tree first, and with enough levels you get very slow planning or a StackOverflowError from the recursive walk.

The rule of thumb: a handful of withColumn calls is fine and more readable; dozens should become one select. If planning time is a noticeable fraction of your job, count the withColumn calls first.

Scope and lifetime.

df.createOrReplaceTempView("trips")              # this SparkSession only
df.createGlobalTempView("trips_global")          # all sessions in the app
spark.sql("SELECT * FROM global_temp.trips_global")

Neither persists metadata: both disappear when the process does. For something that survives, you need a catalog table via saveAsTable, which writes to the metastore.

The reason global temp views exist is multi-session applications. newSession() gives an isolated SQL session sharing the same SparkContext, which is how the Thrift server isolates users. A global temp view is how you deliberately share something across those sessions.

Worth noting: a view stores the query, not the result. Reading a view re-executes its plan. If you want the result stored, cache it or write a table.

A user-defined function: your own logic registered as a callable expression.

@F.udf("double")
def add_fee(fare):
    return fare * 1.1 if fare else None

df.withColumn("total", add_fee("fare_amount"))

Avoid it when a built-in exists, for three compounding reasons:

1. Catalyst cannot see inside it. It is a black box, so no predicate can be pushed through it, no constant folded, no expression simplified.
2. It breaks whole-stage codegen across that point, so the tight generated loop becomes iterator-by-iterator.
3. In PySpark it leaves the JVM. Rows are serialized to a Python worker and results serialized back, per row for a plain UDF.

Almost everything people write a UDF for exists already:

# instead of a UDF
F.when(F.col("fare") > 100, "high").otherwise("mid")
F.regexp_extract("path", r"city_id=(\w+)", 1)
F.coalesce("a", "b", F.lit(0))
F.date_format("started_at", "yyyy-MM-dd")

The ordering to remember: built-in functions, then SQL expressions, then a pandas (vectorised) UDF, then a Scala UDF if you are on the JVM, and a plain Python UDF last. If you must write one, keep it narrow and apply it after filtering, so it sees as few rows as possible.

A vectorised UDF that receives an Arrow batch as a pandas Series and returns one, rather than being called per row.

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def adjust(fare: pd.Series) -> pd.Series:
    return fare * 1.1

The cost of a Python UDF is dominated by crossing the JVM-Python boundary and by per-row interpreter overhead. A pandas UDF amortises the crossing over a whole batch and lets the work happen in vectorised NumPy or pandas code, so the per-row cost falls sharply.

It still breaks whole-stage codegen, because the data still leaves the JVM. So the ordering is: built-ins beat pandas UDFs, and pandas UDFs beat plain Python UDFs.

Variants worth knowing:

• Series to Series, the scalar case above.
• Iterator of Series, when you need expensive per-partition setup such as loading a model once.
• Grouped map, via df.groupBy(...).applyInPandas(fn, schema), which hands you a whole group as a DataFrame. Watch memory: one group must fit.
• mapInPandas, which streams batches of an entire DataFrame without grouping.

Arrow is what makes this work, and it is also what makes toPandas() faster when enabled.

Use null-aware operators, because in SQL semantics any comparison with null is null, not false.

df.filter(F.col("city_id").isNull())
df.filter(F.col("city_id").isNotNull())
df.filter(F.col("a").eqNullSafe(F.col("b")))     # null == null is true here

df.na.fill({"fare_amount": 0.0, "city_id": "unknown"})
df.na.drop(subset=["trip_id"])
df.withColumn("city", F.coalesce("city_id", F.lit("unknown")))

The traps that produce silently wrong results:

• col == value never matches nulls. A filter meant to exclude one city also drops every row whose city is null.
• An inner join drops null keys entirely, because null never equals null.
• Aggregations skip nulls. count("col") counts non-null values while count("*") counts rows, and avg ignores nulls rather than treating them as zero.
• A large number of null keys is skew, because they all hash to one partition.

And one version-specific change to know: from Spark 4.0 spark.sql.ansi.enabled is on by default, so invalid casts and overflows now raise instead of quietly producing null. That converts a class of silent null-generation into a visible error, which is usually an improvement even when it breaks a job on upgrade.

It became the default. The migration guide is explicit:

Since Spark 4.0, spark.sql.ansi.enabled is on by default.

What changes with it on:

To restore the old behaviour:

spark.conf.set("spark.sql.ansi.enabled", "false")

There is also a SPARK_ANSI_SQL_MODE environment variable, which the source reads when computing the default.

The right framing for an interview is that this converts silent data corruption into loud failure. Under the old behaviour a malformed value became null, flowed through the pipeline, and quietly changed a downstream aggregate. Under ANSI the job fails and you find out.

So when a job breaks on upgrade, treat it as a finding rather than a regression: something in the data was already not what the schema claimed. Fix the cast explicitly with try_cast where null genuinely is the right answer, rather than disabling ANSI globally.

union matches columns by position. unionByName matches them by name.

a = spark.createDataFrame([(1, "sf")], ["id", "city"])
b = spark.createDataFrame([("nyc", 2)], ["city", "id"])

a.union(b).show()          # id=nyc, city=2   <- silently wrong
a.unionByName(b).show()    # correct

This is one of the few Spark behaviours that produces silently wrong data rather than an error, because the types happened to be compatible. If the types had differed you would have got an analysis error and noticed.

unionByName also handles schema differences:

a.unionByName(b, allowMissingColumns=True)   # missing columns become null

Without that flag, a name present on one side and not the other is an error, which is usually what you want.

The recommendation is to default to unionByName, and to reach for union only when you have deliberately aligned the schemas and want positional semantics. In a code review, a bare union on two DataFrames built by different code paths is worth questioning.

Note also that both are narrow transformations: they do not shuffle, they just concatenate partitions. Neither deduplicates, unlike SQL’s UNION, which is union followed by distinct.

partitionBy writes a directory per distinct value, which lets later readers prune whole directories.

(df.write
   .partitionBy("city_id")
   .mode("overwrite")
   .parquet("s3a://lakehouse-prod/warehouse/trips"))

trips/city_id=sf/part-00000-....parquet
trips/city_id=nyc/part-00000-....parquet

The trap is file count, and it is arithmetic: every task that holds rows for a partition value writes a file into it. With 200 shuffle partitions and 250 cities you can get up to 50,000 files.

The fix is to repartition by the same columns first, so each value is held by one task:

(df.repartition("city_id")
   .write.partitionBy("city_id")
   .mode("overwrite")
   .parquet(path))

Other things to get right:

• Cardinality. Partition on a low-cardinality column. Partitioning by a timestamp or a user id creates a directory per value, which makes listing slower than scanning.
• Dynamic overwrite. spark.sql.sources.partitionOverwriteMode set to dynamic makes overwrite replace only the partitions present in the data, instead of the whole table.
• File size. spark.sql.files.maxRecordsPerFile, default 0 meaning unlimited, caps rows per file when one partition is huge.

Both pre-organise data on write, but they solve different problems.

(df.write
   .bucketBy(64, "rider_id")
   .sortBy("rider_id")
   .mode("overwrite")
   .saveAsTable("lakehouse.trips_bucketed"))

Bucketing works because the bucket count and key are recorded in the metastore, so when two tables are bucketed the same way on the join key, Spark knows matching rows are already co-located and skips the exchange. That is the only way to avoid a shuffle on a large-to-large join.

The conditions are strict, and naming them is a good answer: both sides bucketed, on the same column, with the same number of buckets. Mismatched bucket counts means Spark shuffles anyway.

The usual pattern is to combine them: partitionBy a date for pruning, and bucketBy a join key within each partition.

Four, set with .mode():

df.write.mode("append").parquet(path)

The two that need care:

overwrite on a partitioned table replaces the whole table by default, even if your DataFrame only contains one day. Setting

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

makes it replace only the partitions present in the data, which is almost always what an incremental job wants.

append has no transactional guarantee on plain Parquet. Two concurrent appends can interleave, and a failure mid-write leaves partial files with no way to roll back, because a directory of Parquet files has no commit protocol. That is the structural reason lakehouse formats exist: Hudi, Iceberg and Delta add atomic commits over the same files.

ignore is worth flagging in review, because it silently does nothing, which hides a logic error rather than surfacing it.

Three properties that compound:

1. Columnar. Columns are stored separately, so reading 3 of 200 columns touches roughly 1.5% of the bytes. Row formats must read and parse the whole row.
2. Statistics per row group. Each row group carries min, max and null counts per column, so a pushed-down predicate can skip whole groups without decoding them.
3. Compression that works. A column holds like-typed, often similar values, so dictionary and run-length encoding are very effective, which reduces both storage and read time.

It also carries a schema, including nested structures, so there is no inference step and no ambiguity about types.

spark.read.parquet(path)     # schema comes from the files

spark.sql.parquet.filterPushdown defaults to true, so the statistics are used by default.

When it is not the right choice:

• Write-once-read-never logs, where a cheaper row format is fine.
• Row-level random access, which no scan format does well.
• Very small datasets, where the per-file overhead dominates.
• Interchange with tools that cannot read it, where CSV wins on ubiquity alone.

ORC is broadly comparable and stronger in some Hive-centric stacks; the columnar argument is the same. Avro is row-oriented and better for streaming records and schema evolution on the write path, which is why Hudi uses it for log blocks.

Give Spark a column to split on, plus bounds and a partition count. Without them the entire table is read by a single task, whatever your cluster size.

trips = (spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://db.internal:5432/rides")
    .option("dbtable", "public.trips")
    .option("user", "etl")
    .option("partitionColumn", "trip_seq")
    .option("lowerBound", "1")
    .option("upperBound", "40000000")
    .option("numPartitions", "16")
    .option("fetchsize", "10000")
    .load())

Spark turns this into 16 queries with disjoint ranges on trip_seq.

The rules:

• partitionColumn must be numeric, date or timestamp. A string primary key will not work, which catches people with UUID keys.
• All four options go together. Setting numPartitions alone does nothing.
• The bounds are for splitting, not filtering. Rows outside them are still read, in the first and last partitions, so badly chosen bounds create skew rather than dropping data.
• fetchsize controls JDBC round trips and often matters more than partitioning for a moderate table.

And the caution to volunteer: 16 parallel connections is 16 concurrent queries against a production database. Parallelism here is a load decision on the source system, not just a Spark setting.

They do unrelated things. Caching stores a computed result; a view stores a query.

df.createOrReplaceTempView("trips")   # names a plan
df.cache()                            # stores the data
spark.sql("CACHE TABLE trips")        # names a plan AND stores the data

So reading a temporary view re-executes its plan every time. If the plan is expensive and you read it repeatedly, you want caching as well, which is what CACHE TABLE does.

One behavioural difference worth knowing: CACHE TABLE is eager by default, populating immediately, while df.cache() is lazy and populates on the next action. CACHE LAZY TABLE makes the SQL form lazy.

Two related notes:

• Caching is by plan, not by name. Two DataFrames with identical plans share a cache entry, which is why caching then re-deriving the same query can hit the cache unexpectedly.
• A cached DataFrame is pinned until released. unpersist() or spark.catalog.clearCache(); otherwise it holds storage memory for the life of the application and squeezes execution memory from every later stage.

Ten steps, and being able to narrate them in order is the answer.

flowchart TB
  A["<b>Startup</b> (1-4)<br/>submit, driver starts,<br/>executors requested and registered"]
  B["<b>Plan</b> (5-7)<br/>code builds a plan, an action submits a job,<br/>Catalyst optimises, DAGScheduler cuts stages"]
  C["<b>Execute</b> (8-9)<br/>TaskSchedulerImpl places tasks,<br/>executors run them and write shuffle output"]
  D{"more stages?"}
  E["<b>Finish</b> (10)<br/>results to the driver,<br/>or written by the executors"]
  A --> B --> C --> D
  D -->|"yes"| C
  D -->|"no"| E

Two details that make the answer stronger:

• Step 5 is where laziness lives. Everything between the session and the first action is plan construction, so a job that fails at step 6 may have a bug written many lines earlier.
• Step 10 forks. collect() pulls every row back to the driver; write has each executor write its own partition and return only a status. Same job shape, completely different driver cost.

Steps 8 and 9 repeat per stage, and that loop is where adaptive query execution intervenes, because between two passes it has real statistics from the stage that just finished.

Nine responsibilities, in roughly the order they happen:

1. Creates the SparkSession, bringing up SparkContext, SparkEnv, the block manager and the RPC endpoints executors call back on.
2. Asks for executors. SchedulerBackend requests resources from whichever cluster manager --master named.
3. Turns code into a plan. Unresolved, then analysed against the catalog, then optimised by Catalyst, then a physical plan chosen by cost.
4. Cuts the plan into stages. DAGScheduler walks the RDD lineage and breaks it at every ShuffleDependency, producing a TaskSet per stage.
5. Places every task. TaskSchedulerImpl and TaskSetManager decide which executor runs which task, preferring the best available locality.
6. Tracks every attempt. Retries up to spark.task.maxFailures, default 4; speculates slow tasks if enabled; resubmits stages whose shuffle output was lost.
7. Keeps the block index. BlockManagerMaster knows where every cached partition and shuffle block lives, so executors can fetch from each other.
8. Receives results, bounded in total by spark.driver.maxResultSize, default 1g.
9. Serves the UI on port 4040 and writes the event log the history server replays.

The two facts that follow: spark.driver.memory defaults to 1g, which is a coordinator’s budget and not a data-processing one; and the driver is the one process with no redundancy, because items 3 to 7 exist nowhere else.

Much less than the driver, deliberately.

1. Registers. CoarseGrainedExecutorBackend sends RegisterExecutor and waits for RegisteredExecutor before accepting work. Until that round trip completes, a granted container is idle.
2. Accepts LaunchTask and runs each task in a thread from its pool. Concurrency is spark.executor.cores, which defaults to 1 on YARN and to all cores in standalone mode.
3. Reads input, runs the generated code, and on a shuffle map stage writes partitioned output to local disk.
4. Reports back with StatusUpdate per task and a heartbeat every spark.executor.heartbeatInterval, default 10s. Miss enough and the driver declares it lost.
5. Stores and serves blocks through its BlockManager, which makes a cached DataFrame reusable and lets the next stage fetch shuffle data from peers.

The contract is narrow on purpose: an executor never makes a scheduling decision, never sees the query plan, and never coordinates with another executor about what to run. It receives tasks, runs them, serves bytes.

That is exactly what makes executors disposable. Lose one and its tasks are retried elsewhere with its cached partitions recomputed from lineage. It is also why executors are per-application rather than shared: memory and slots are reserved for one application’s work.

One thing: decide who gets containers. It is a resource manager, not a work scheduler.

Kubernetes is the clearest illustration of the split, because there is no Spark cluster-manager daemon at all: the driver talks to the API server directly and asks for pods. The component granting resources and the component scheduling work are simply different things, however they are packaged.

Two defaults follow from this being a separate concern: spark.dynamicAllocation.enabled is false, so absent a decision your application holds its executors from first request to exit, idle or not. And Mesos was removed in the 4.x line, so the three rows above are the whole list.

The practical consequence people find surprising: tuning YARN’s scheduler changes what you are given, never how your tasks are distributed across what you were given. That is the driver’s job.

Stage-level scheduling. Given a job, it:

1. Walks the RDD lineage backwards from the action.
2. Cuts a new stage at every ShuffleDependency, absorbing all narrow dependencies into the current stage.
3. Determines the order stages must run in, and which can run in parallel.
4. Submits each ready stage to TaskScheduler as a TaskSet.
5. Handles FetchFailedException by resubmitting the upstream stage, because lost shuffle output cannot be re-fetched.
6. Computes preferred locations for tasks from the RDD’s getPreferredLocations.

The distinction from TaskScheduler is the point of the question: DAGScheduler decides what stages exist and in what order; TaskSchedulerImpl decides which executor runs which individual task. One is about the shape of the job, the other about placement.

flowchart TB
  J["job"] --> D["DAGScheduler<br/>cuts stages at shuffles"]
  D --> TS["TaskSet per stage"]
  TS --> T["TaskSchedulerImpl<br/>places each task"]
  T --> B["SchedulerBackend"] --> E["executors"]

The stage-resubmission behaviour is the one to volunteer, because it explains a confusing UI: a job that appears to go backwards, re-running a stage that had already completed, is DAGScheduler rebuilding shuffle output that died with an executor.

Task-level scheduling: taking a TaskSet and deciding which executor runs each task in it.

Its concerns:

• Placement by locality. It offers each task the best available locality level, waiting briefly (spark.locality.wait) before settling for a worse one.
• Retries. Failed tasks are rescheduled up to spark.task.maxFailures, default 4, preferring a different executor when the failure looked executor-specific.
• Speculation. If enabled, duplicating tasks that lag their peers.
• Excluding bad executors. If one executor or node keeps failing tasks, it can stop scheduling there so a single bad machine does not burn every task’s retry budget.
• Fair or FIFO ordering between concurrent TaskSets, per spark.scheduler.mode, default FIFO.

It works with a TaskSetManager per stage, which holds the per-task state, and talks to the cluster through SchedulerBackend.

The reason to know the split: when tasks are not being placed where you expect, the question is a TaskSchedulerImpl and locality question. When stages are not in the shape you expect, it is a DAGScheduler question. Knowing which component owns which symptom is most of debugging.

The single component that talks to the cluster manager, with an implementation per manager. It is the boundary between Spark’s scheduling and the outside world.

Its job:

• Request and release executors. Including the dynamic-allocation calls to scale up and down.
• Receive executor registration and tell the scheduler how many slots exist.
• Launch tasks onto executors, and relay their status updates back.

Implementations include CoarseGrainedSchedulerBackend as the common base, with YARN, Kubernetes and standalone specialisations, plus LocalSchedulerBackend for local[*].

“Coarse-grained” is worth explaining because the name is opaque: the executor is acquired once and held for the application’s lifetime, running many tasks. The alternative, fine-grained acquisition per task, existed under Mesos and is gone.

This is the layer that makes Spark portable. The whole scheduler above it is identical on every cluster manager, and only SchedulerBackend changes. That is why the same jar runs on YARN and Kubernetes with nothing but --master different.

Every stage but the last in a job is a ShuffleMapStage; the last is a ResultStage.

The reason the distinction exists in the scheduler is recovery. ShuffleMapStage output is registered with MapOutputTracker on the driver, so Spark knows where every map output block lives. If an executor dies and takes blocks with it, DAGScheduler knows which stage to resubmit and which partitions of it are missing.

A ResultStage has nothing registered, because its output goes to the driver or to storage.

This is also why the UI shows the final stage differently, and why a ShuffleMapStage can be skipped: if its output is still registered from a previous job, Spark reuses it rather than recomputing. Those grey “skipped stages” in the UI are a feature, not a problem, and they are what makes reusing a shuffle across two actions cheap.

Five, best first:

TaskSchedulerImpl offers each task the best level available, and if nothing at that level is free it waits spark.locality.wait, default 3s, before settling for the next level down.

flowchart LR
  A["task ready"] --> B{"PROCESS_LOCAL<br/>slot free?"}
  B -->|"yes"| R["run"]
  B -->|"no"| C["wait spark.locality.wait"]
  C --> D{"NODE_LOCAL free?"}
  D -->|"yes"| R
  D -->|"no"| E["degrade further, to ANY"]

The modern caveat is the useful part. Locality was designed for HDFS, where compute ran on the machines holding the blocks. Reading from S3 or another object store means no partition has a meaningful preferred location, so most tasks are NO_PREF or ANY and the waiting buys nothing. On such clusters, lowering spark.locality.wait can measurably reduce scheduling delay.

Where locality still matters is cached data: a PROCESS_LOCAL hit on a cached partition avoids a network fetch entirely.

Not from the number of files, which is the assumption that costs people the most time.

FilePartition.maxSplitBytes decides the split size first:

Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))

• defaultMaxSplitBytes is spark.sql.files.maxPartitionBytes, default 128MB.
• openCostInBytes is spark.sql.files.openCostInBytes, default 4MB, charged per file so many small files are packed together rather than each getting a task.
• bytesPerCore is the total size divided by spark.sql.files.minPartitionNum, which falls back to the default parallelism.

The consequence is measurable. Write the same data as 1, 8 and 16 Parquet files, then read each back:

File count did not determine partition count in any of them, because the files were small enough to pack.

Two directions follow: a directory of small files collapses into far fewer partitions than it has files, and a single file larger than the split size is divided into several. So read the number rather than predicting it:

print(df.rdd.getNumPartitions())

One per partition of that stage. Nothing else affects it.

Where that partition count comes from depends on the stage:

• The first stage takes it from input splitting, so spark.sql.files.maxPartitionBytes and file sizes.
• A stage after a shuffle takes it from spark.sql.shuffle.partitions, default 200, or whatever adaptive coalescing reduced it to at runtime.
• After an explicit repartition(n), it is n.

It is not related to the number of executors, and that confusion is the reason the question gets asked. With 200 tasks and 10 executors of 4 cores each, you have 40 slots, so the stage runs in 5 waves. Tasks queue for slots; they are not divided among executors up front.

The arithmetic worth having ready:

slots = num_executors x spark.executor.cores
waves = ceil(tasks / slots)

A stage with fewer tasks than slots leaves cores idle. A stage with thousands of very short tasks spends its time in scheduling. The target is several waves of tasks each doing a meaningful amount of work, which is what “partitions in the low hundreds of megabytes” is really a proxy for.

The work is redone, and the reason it can be redone is lineage.

Recovery is per partition, not per job: losing one executor costs the partitions it held, which are rebuilt on whichever executors remain while everything else stands. Its running tasks are retried elsewhere, counting against spark.task.maxFailures.

Cached blocks are recomputed from lineage, unless the storage level replicated them (MEMORY_ONLY_2 and friends), in which case the replica is promoted.

Shuffle output is the special case worth volunteering:

flowchart LR
  A["stage 0 completed"] --> B["executor dies with<br/>its shuffle files"]
  B --> C["stage 1 fetch fails<br/>FetchFailedException"]
  C --> D["DAGScheduler resubmits<br/><b>stage 0</b>"]

That is a stage rerun rather than a task retry, and it is why a job can appear to go backwards in the UI. An external shuffle service avoids it by serving shuffle blocks from a node-level daemon that outlives the executor, which is also why dynamic allocation wants one.

The contrast with the driver is the closing point: executors hold only recomputable state, so they are disposable; the driver holds the plan, the scheduler and the block index, so it is not.

Because of what each holds that exists nowhere else.

An executor’s contents are either derivable (cached partitions, from lineage) or reproducible (shuffle output, by rerunning the stage). So Spark can always reconstruct them.

The driver’s contents are not derivable from anything. It is the only process that knows what the job is, which stages remain, which tasks succeeded, and where every block lives. There is no second copy, so there is nothing to recover from.

What you can do about it:

• Cluster deploy mode puts the driver inside the cluster, where the manager can restart it. On standalone, --supervise restarts a failed driver.
• Structured Streaming checkpoints let a restarted driver resume from recorded offsets and state, which is the closest thing to driver fault tolerance in practice.
• Keep the driver unloaded. Most driver failures are self-inflicted: collect() on a large result, a broadcast that is not small, or hundreds of thousands of tasks to track.

A daemon running on each node that serves shuffle blocks on behalf of the executors that wrote them. spark.shuffle.service.enabled defaults to false.

Without it, shuffle output lives inside the executor process, so the executor must stay alive for as long as anything might fetch from it. With it, the blocks are on the node’s disk and served by the daemon, so the executor can exit freely.

flowchart TB
  subgraph N["without the service"]
    E1["executor<br/>holds shuffle files"] -->|"must stay alive"| F1["reduce tasks fetch"]
  end
  subgraph W["with the service"]
    E2["executor<br/>can exit"] --> D2["node shuffle service<br/>serves the files"] --> F2["reduce tasks fetch"]
  end

Why it matters:

• Dynamic allocation needs it, or an equivalent. Otherwise Spark cannot release an idle executor that still holds shuffle data someone might want. The alternative is spark.dynamicAllocation.shuffleTracking.enabled, default true, which tracks which executors hold live shuffle blocks and keeps only those.
• It reduces FetchFailedException, because an executor dying no longer destroys its shuffle output.
• It offloads serving work from the executor JVM, which is otherwise doing your computation.

The cost is an extra daemon to deploy and operate per node, which on YARN means an auxiliary service in the NodeManager.

Adding and removing executors according to pending work, rather than holding a fixed set.

spark.dynamicAllocation.enabled defaults to false, so absent a decision your application holds its executors from first request to exit, idle or not.

How it behaves:

• Scales up when tasks are pending, adding executors exponentially up to maxExecutors.
• Scales down when an executor has been idle for executorIdleTimeout.
• Respects minExecutors as a floor, and initialExecutors at start.

--conf spark.dynamicAllocation.enabled=true --conf spark.dynamicAllocation.minExecutors=2 --conf spark.dynamicAllocation.maxExecutors=100

The shuffle problem is the part to know. An idle executor may still hold shuffle blocks a later stage needs, so releasing it would cause fetch failures. Two solutions: an external shuffle service, or spark.dynamicAllocation.shuffleTracking.enabled, which defaults to true and keeps executors that hold live shuffle data.

When it helps: interactive and bursty workloads, notebooks, multi-tenant clusters, and jobs whose parallelism varies a lot between stages.

When it hurts: short jobs, where the ramp-up costs more than it saves, and jobs with strict latency needs, where waiting for executors is worse than holding them. Note also that setting --num-executors alongside it is contradictory: the static value only acts as an initial size.

Re-running tasks that are running much slower than their peers, and taking whichever attempt finishes first. spark.speculation defaults to false.

Spark watches the distribution of task durations within a stage. Once a configurable fraction of tasks has completed, any task running more than spark.speculation.multiplier times the median is duplicated on another executor.

When it helps: a genuinely slow machine. A failing disk, a noisy neighbour, a node with thermal throttling. The duplicate runs elsewhere and finishes normally.

When it hurts, and this is the answer interviewers want: data skew. If the task is slow because its partition holds ten times the rows, the duplicate has exactly the same work to do and is equally slow. You have now consumed two slots to get one result no sooner. Worse, it can mask the skew, because the symptom looks like flaky infrastructure rather than an uneven distribution.

So the diagnostic order is: check whether slow tasks also have larger shuffle read. If they do, it is skew and speculation is the wrong tool. If durations vary while input sizes are equal, it is the environment and speculation helps.

One more caution: with a non-idempotent sink, two attempts writing the same output can duplicate data. Speculation assumes writes are safe to repeat.

spark.scheduler.mode defaults to FIFO.

That governs how concurrent jobs within one application share executor slots, which only matters when you submit jobs from multiple threads or serve multiple users from one SparkSession.

Under FIFO, a large job submitted first can starve a small interactive query submitted second, because it holds every slot until it is done. That is precisely the wrong behaviour for a notebook server or a Thrift server, where a quick query should not wait behind an hour-long batch.

--conf spark.scheduler.mode=FAIR --conf spark.scheduler.allocation.file=/etc/spark/fairscheduler.xml

FAIR also supports pools with weights and minimum shares, so you can give an interactive pool priority over a batch pool. A job joins a pool by setting a thread-local property:

sc.setLocalProperty("spark.scheduler.pool", "interactive")

Keep the levels distinct: this is scheduling within an application. Sharing between applications is the cluster manager’s job, with YARN queues or Kubernetes quotas.

Two configs, one per side:

spark-submit \
  --conf "spark.driver.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties" \
  --conf "spark.executor.extraJavaOptions=-Dlog4j.configurationFile=log4j2.properties -verbose:class" \
  app.jar

They are the delivery mechanism for almost every JVM-level change: logging configuration, -verbose:class for class-loading problems, -Xss for stack size, heap dump flags, proxy properties, and GC settings.

The client-mode trap is the part that catches people. In client mode the driver JVM is already running by the time your SparkConf executes, so a driver JVM option set in code arrives too late to have any effect. Use --driver-java-options, or a properties file, or set it before the JVM starts.

Two more notes:

• Do not set -Xmx here. Heap comes from spark.driver.memory and spark.executor.memory; setting it in extra options conflicts with them.
• Avoid secrets. The Environment tab of the UI displays configs, so a password in extraJavaOptions is visible to anyone who can reach the UI.

To confirm a flag took effect, the Environment tab is authoritative, and --verbose on submit prints where each value came from.

A cap on the total serialized size of results returned to the driver, default 1g. Exceeding it aborts the job with an error naming the config.

It exists because without it, collect() on a large dataset would simply exhaust the driver heap and produce an OutOfMemoryError, which is harder to diagnose and can take the application down mid-flight. The cap converts that into a clear, early, named failure.

spark.conf.get("spark.driver.maxResultSize")     # 1g

Note it counts the total across all tasks in an action, not per task, and it is checked as results arrive, so a job can fail partway through collecting.

The right response is almost never to raise it. If you are hitting a 1 GB limit, the question is why a gigabyte of rows needs to be in the driver:

• Writing out? Use write, so executors write in parallel and only statuses return.
• Inspecting? show(), take() or limit().
• Counting or aggregating? Do it as an aggregation, which returns one row.
• Handing to Python? toPandas() has the same problem; consider mapInPandas to keep the work distributed.

Setting it to 0 disables the check, which trades a clear error for an OutOfMemoryError. Raising it is defensible only when you genuinely need a large result on the driver and have sized spark.driver.memory accordingly.

spark.ui.port defaults to 4040. If that port is taken, Spark increments and tries 4041, 4042 and so on, which is why a second concurrent application appears on a different port.

The UI dies with the application. It is served by the driver, so when the driver exits there is nothing left to serve it. That is a problem, because most investigation happens after a job has failed or finished.

The solution is two-part:

--conf spark.eventLog.enabled=true --conf spark.eventLog.dir=hdfs:///spark-events

The driver writes an event log, and a separately running history server reads that directory and replays the logs, giving you the same UI for completed applications.

Without event logging, a job that failed overnight leaves you with only the driver’s stdout, which is why enabling it is close to mandatory in production.

Two practical notes: event logs grow, so they need retention and ideally compression (spark.eventLog.compress); and on YARN the ResourceManager UI links to the history server, so the “tracking URL” for a finished application lands there rather than at a dead port 4040.

They are the two sanctioned ways to move data between driver and executors, one in each direction.

A task runs in another JVM, so an ordinary variable captured by a closure is serialized, shipped and then discarded. Updates to it never come back, which is why both mechanisms exist.

flowchart LR
  D["driver"] -->|"broadcast: one copy per executor"| E["executors"]
  E -->|"accumulator: merged updates"| D

Broadcast, outward. A read-only value cached once per executor rather than serialized per task. The saving is per-task-count: with 10,000 tasks and a 10 MB table, one copy per executor instead of 10,000 copies.

Accumulators, inward. Tasks add, the driver reads. The guarantee is exactly-once only for updates inside actions; in a transformation, a recompute or retry applies the update again.

What neither does: give you shared mutable state. There is no way for two tasks to coordinate through a variable, which is deliberate. If tasks need to agree on something, that is what a shuffle is for.

The common mistake is closing over a large object without broadcasting it, which silently multiplies serialization cost by the task count and often shows up as inexplicably slow task launch.

A client-server architecture where the client builds an unresolved logical plan and sends it to a remote Spark server over gRPC, instead of the client being the driver.

Traditionally your application is the driver: it holds the SparkContext, it must be a JVM (or a JVM plus a Python process), and it must be network-close to the executors. Spark Connect splits that.

flowchart LR
  C["thin client<br/>Python, Scala, Go, JVM"] -->|"unresolved plan<br/>over gRPC"| S["Spark Connect server<br/>holds the real driver"]
  S --> E["executors"]

What it buys:

• Thin clients. No local JVM, no cluster-side dependencies, so a notebook, an IDE or a web service can drive a cluster.
• Isolation. User code crashing or leaking memory no longer takes down the driver, because it is not in the driver.
• Upgradeability. Client and server versions are decoupled, so the cluster can move independently.
• Multi-language. The protocol, not the JVM, is the interface, which is how non-JVM clients become possible.

The trade is that anything depending on being inside the driver does not work: SparkContext internals, RDD APIs in general, and code that assumes JVM colocation. So it is a DataFrame and SQL interface, which covers most modern usage.

A redistribution of data across partitions so that rows which must be processed together end up in the same partition.

It has two halves:

1. The map side writes each record into a bucket for its target partition, sorts or spills as needed, and writes partitioned output to local disk, along with an index of where each partition’s bytes start.
2. The reduce side fetches its own slice from every map output, over the network, then merges them.

flowchart TB
  subgraph M["map side, stage 0"]
    M1["task 0"] --> F1[("file + index")]
    M2["task 1"] --> F2[("file + index")]
    M3["task 2"] --> F3[("file + index")]
  end
  subgraph R["reduce side, stage 1"]
    F1 --> R1["task 0 fetches<br/>its slice from all 3"]
    F2 --> R1
    F3 --> R1
    F1 --> R2["task 1"]
    F2 --> R2
    F3 --> R2
  end

Three consequences worth stating:

• It always touches disk, even when nothing spills, because map output is written to local disk before being served. Shuffle is not an in-memory operation.
• It is all-to-all. With m map tasks and r reduce tasks there are m x r block fetches, which is why very high parallelism on both sides is its own cost.
• It creates a barrier. Stage 1 cannot start until stage 0 has finished writing every partition, so one straggler delays everything.

Any wide dependency. In practice:

What does not shuffle: map, filter, select, withColumn, union, coalesce, and a join where one side is broadcast.

The reliable way to count them is the plan, because every shuffle is an Exchange:

df.explain("formatted")

Exchange hashpartitioning(city_id#21, 200), ENSURE_REQUIREMENTS

The description tells you the key and the target partition count, so you can see not just that it shuffles but on what.

The optimisation this enables: if two consecutive operations shuffle on the same key, Spark needs only one exchange. So aggregating and joining on the same column is much cheaper than on different columns, and that is often something you can arrange.

Five costs, and naming several of them is the difference between a shallow and a deep answer.

1. Serialization and deserialization. Every record crossing the boundary must be encoded and decoded.
2. Disk I/O. Map output is always written to local disk, and spills add more.
3. Network transfer. Every reduce task pulls from every map task, so the fetch pattern is all-to-all.
4. Sorting or merging. Depending on the writer path, records are sorted on the map side and merged on the reduce side.
5. The stage barrier. No reduce task can start before every map task has finished, so the slowest map task sets the floor.

The barrier is the one people forget, and it is why skew is so damaging: a single slow map task holds up the entire downstream stage, wasting every other slot.

There is also a scaling cost: with m map and r reduce tasks there are m x r blocks to track and fetch. Very high parallelism on both sides turns into metadata and connection overhead, which is one reason “just raise the partition count” has a limit.

What to do about it, in order: eliminate the shuffle (broadcast, or bucket the tables); reduce the data entering it (filter and project first); reduce the number of shuffles (aggregate and join on the same key); then size its partitions sensibly.

SortShuffleManager selects between three writer paths, checked cheapest first:

The condition that disqualifies both fast paths is the same: a map-side combine. And a map-side combine is exactly what reduceByKey, aggregateByKey and a partial aggregation do.

So there is a real tension worth articulating:

reduceByKey   ->  less data shuffled, through the slowest writer
groupByKey    ->  more data shuffled, possibly through a faster writer

reduceByKey still usually wins, because network volume dominates writer efficiency, and because groupByKey risks an OutOfMemoryError on a hot key. But knowing why the trade exists, rather than reciting “always use reduceByKey”, is what a deeper question is probing for.

For DataFrames you rarely choose directly: Catalyst inserts partial aggregation automatically, which is a map-side combine, so aggregations take the sort path.

An uneven distribution of rows across partitions, so some tasks have far more work than others.

The cause is almost always the data itself: one key dominates. A city_id column where 60% of trips are in one city, a customer_id where a single enterprise account has millions of rows, or a null key that every unmatched row shares.

flowchart TB
  subgraph E["even"]
    A1["task 0: 1M rows"] --- A2["task 1: 1M"] --- A3["task 2: 1M"] --- A4["task 3: 1M"]
  end
  subgraph S["skewed"]
    B1["task 0: 100K"] --- B2["task 1: 100K"] --- B3["task 2: <b>3.7M</b>"] --- B4["task 3: 100K"]
  end

Why it is so costly: a task is the unit of work and cannot be split. One task with ten times the rows takes roughly ten times as long, and because of the stage barrier, every other slot sits idle waiting for it. A 200-task stage can be limited by one task.

The signatures to recognise:

• One or two tasks with far longer duration than the median, in the same stage.
• Those same tasks showing much larger shuffle read.
• A stage that is 95% complete for a long time.
• An OutOfMemoryError on one executor while others are fine.

Skew is a data problem that presents as a performance problem, which is why config tuning rarely fixes it.

Two places: the UI, and the data.

In the UI, open the stage page and read the task metrics summary, which gives min, 25th percentile, median, 75th and max for duration and shuffle read. A median of 3 seconds with a max of 40 minutes is skew, and the shuffle read column confirms it is data volume rather than a slow machine.

That distinction matters:

In the data, count rows per key and look at the head of the distribution:

(df.groupBy("city_id").count()
   .orderBy(F.desc("count"))
   .show(20, truncate=False))

For an RDD, count rows per partition directly:

print(rdd.mapPartitionsWithIndex(
    lambda i, rows: [(i, sum(1 for _ in rows))]).collect())

And for a join, remember to check both sides plus the null count on the key, since nulls all hash together:

df.filter(F.col("city_id").isNull()).count()

In order of preference, cheapest and least invasive first.

1. Let AQE handle it. spark.sql.adaptive.skewJoin.enabled defaults to true and splits skewed partitions automatically. Check the thresholds first, because they are strict: it needs a partition larger than 5x the median and larger than 256 MB.

2. Broadcast the small side, if there is one. No shuffle means no skew.

trips.join(F.broadcast(cities), "city_id")

3. Salt the hot key. Spread it across N synthetic sub-keys, join, then aggregate the salt away. This is the general answer when the hot key is genuinely large.

4. Separate the hot keys. Process them with a different strategy and union the results. More code, but optimal when there are only a handful.

hot = ["sf"]
a = trips.filter(F.col("city_id").isin(hot)).join(F.broadcast(cities), "city_id")
b = trips.filter(~F.col("city_id").isin(hot)).join(cities, "city_id")
result = a.unionByName(b)

5. Filter nulls out of the join key if they carry no meaning, since they all land in one partition.

What not to do: enable speculation. The duplicate task has the same rows and is equally slow, so it burns a second slot for nothing.

Add a random suffix to the hot key on the fact side, and replicate the dimension side once per possible suffix so every match still exists.

from pyspark.sql import functions as F

N = 16

facts = trips.withColumn("salt", (F.rand() * N).cast("int"))

dims = (cities
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)]))))

joined = facts.join(dims, ["city_id", "salt"]).drop("salt")

flowchart TB
  A["all 'sf' rows<br/>one partition"] --> B["salt 0..15"]
  B --> C["16 partitions:<br/>sf-0, sf-1, ... sf-15"]
  D["cities row 'sf'"] --> E["replicated x16<br/>sf-0 ... sf-15"]
  C --> F["join, now spread<br/>across 16 tasks"]
  E --> F

The trade is explicit: the dimension side is N times larger. With N=16 and a 1 MB dimension that is 16 MB, which is nothing. With a 3 GB dimension it is 48 GB, which is not. So N stays small, and salting suits a large-fact-to-small-dimension shape.

For an aggregation rather than a join, salting is two-phase and cheaper, because nothing is replicated:

(trips
  .withColumn("salt", (F.rand() * N).cast("int"))
  .groupBy("city_id", "salt").agg(F.sum("fare_amount").alias("part"))
  .groupBy("city_id").agg(F.sum("part").alias("total")))

Aggregate per salted key, then aggregate the partials. The first pass spreads the hot key over N tasks; the second is tiny.

Both conditions must hold. A partition is treated as skewed only when it is larger than 5 times the median and larger than 256 MB.

That conjunction is the answer to “AQE skew handling is on but nothing happened”. On a job whose partitions are all under 256 MB, skew handling never fires no matter how uneven they are. A stage with a 2 MB median and a 100 MB outlier is 50x skewed and completely untouched.

# make it fire on a smaller job
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "32MB")

When it does fire, Spark splits the skewed partition into several smaller reduce tasks and replicates the matching partition from the other side so each split still finds its matches. That is salting, done automatically at runtime.

Two limits worth knowing: it applies to joins, so a skewed groupBy is not covered, and it needs the shuffle to have actually happened, so it cannot help a skewed read.

Re-planning parts of a query at runtime, using real statistics from stages that have already finished, instead of relying on planning-time estimates.

It is on by default since Spark 3.2: spark.sql.adaptive.enabled is true.

Three things it does:

flowchart TB
  S["a shuffle stage completes:<br/>real sizes are now known"]
  S --> A["<b>coalesce partitions</b><br/>merge the small ones"]
  S --> B["<b>switch join strategy</b><br/>sort-merge to broadcast"]
  S --> C["<b>split skewed partitions</b><br/>one big task into several"]

Why it works so well: planning-time estimates on derived data are usually wrong. Spark can estimate the size of a table it has statistics for, but after three filters and a join, the estimate is a guess. After the shuffle, it is a measurement.

Reading it: before execution the plan shows AdaptiveSparkPlan isFinalPlan=false, so explain() gives you the planned shape. To see what actually ran, look at the SQL tab after the query finishes.

The main caveat is that AQE works at shuffle boundaries, so a query with no shuffle gets nothing from it, and the first stage is always planned blind.

After a shuffle, it merges small output partitions into fewer, larger tasks.

The relevant settings:

Spark writes the full spark.sql.shuffle.partitions worth of output, looks at the actual size of each, and then assigns contiguous runs of them to single reduce tasks so each task gets roughly the advisory size.

200 shuffle partitions, 800 MB total
   -> without coalescing: 200 tasks x 4 MB
   -> with coalescing:     13 tasks x ~64 MB

This is what makes the static default tolerable. Before AQE, 200 had to be tuned per job or you paid for it; now a high value is mostly harmless, because coalescing brings it down.

The practical consequence for tuning: set the advisory size, not the partition count.

spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

That expresses the intent (“I want partitions about this big”) and lets Spark do the arithmetic per stage, which is better than one number for a query whose stages differ in size.

Note it only merges; it never splits. A stage whose partitions are too large needs a higher spark.sql.shuffle.partitions, or skew handling.

Work from two constraints and take whichever is larger.

Constraint 1, partition size. Aim for partitions in the low hundreds of megabytes of shuffled data. Too large and you spill; too small and scheduling dominates.

partitions = total shuffle bytes / target partition size
           = 600 GB / 128 MB
           = ~4,800

Constraint 2, core count. You want a small multiple of your total cores, so every core gets several tasks and stragglers average out.

cores = num_executors x spark.executor.cores = 50 x 4 = 200
partitions = 200 x 3 = 600      (a floor, not a target)

With AQE, the better move is to express the first constraint directly and let coalescing handle the rest:

spark.conf.set("spark.sql.shuffle.partitions", "2000")            # generous
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

Set the count high enough that no partition is too big, and let AQE merge the small ones down. That handles a query whose stages differ in volume, which a single static number cannot.

The diagnostic loop: if the stage page shows spill, partitions are too large. If it shows thousands of tasks lasting milliseconds, they are too small. Those two symptoms are the feedback signal, not a formula.

Many tiny files make every later read slow, for reasons that are not about total bytes:

• Listing cost. Planning must enumerate the files, which on object storage means many API calls.
• Per-file open cost. Spark charges spark.sql.files.openCostInBytes, default 4MB, per file when computing splits, precisely because opening one is not free.
• Metadata overhead. Each Parquet file has a footer to read.
• Compression loss. Small files compress worse, because dictionaries and encodings have less to work with.

Where they come from, and the fix for each:

(df.repartition("city_id")
   .write.partitionBy("city_id")
   .mode("overwrite").parquet(path))

You can also cap file size rather than count:

spark.conf.set("spark.sql.files.maxRecordsPerFile", "5000000")

which defaults to 0, meaning unlimited.

The structural fix for a continuously written table is a lakehouse format, because Hudi, Iceberg and Delta all provide compaction as a managed table service rather than something you bolt on.

repartition(n) distributes rows round-robin into n partitions, producing even sizes and no relationship between a row’s value and its partition. repartition(col) hash-partitions by that column, so all rows sharing a value land together.

df.repartition(400)                 # even, value-agnostic
df.repartition("city_id")           # by value, into spark.sql.shuffle.partitions
df.repartition(400, "city_id")      # by value, into 400

The catch with repartition(col) is skew. Hash partitioning a column where one value dominates puts all those rows in one partition, by design. So it fixes file layout and can remove a later shuffle, while potentially creating the very skew you were trying to avoid. Check the distribution before reaching for it.

The most common correct use is immediately before a partitioned write, so each output directory is produced by one task rather than all of them. The second is before several operations that key on the same column, so one exchange serves them all instead of each operator shuffling separately.

Yes, and it is more expensive than most shuffles because of how it partitions.

A global sort needs range partitioning: partition 0 gets the lowest values, partition 1 the next, and so on, so that concatenating the sorted partitions yields a globally sorted result. But Spark does not know the value distribution in advance, so it must sample the data first to pick range boundaries.

flowchart LR
  A["sample the data<br/><i>an extra job</i>"] --> B["compute range boundaries"]
  B --> C["shuffle by range"]
  C --> D["sort within each partition"]

That sampling job is why orderBy can show an unexpected extra job in the UI before the real work starts.

If you only need ordering within partitions, which is often enough (for example to make a file’s contents ordered, or for a window already partitioned by the same key), use the cheaper form:

df.sortWithinPartitions("started_at")     # no range partitioning, no sampling

Two related notes:

• A RangePartitioner handles skew badly when many rows share a value, since boundaries cannot split a single value.
• orderBy before a limit is a special case Spark optimises with TakeOrderedAndProject, which avoids a full global sort by taking the top n per partition and merging.

Three ways, and each removes the shuffle for a different reason.

1. Broadcast the small side. Every executor gets the whole small table, so each large-side partition can join locally. No redistribution needed.

trips.join(F.broadcast(cities), "city_id")

Automatic when the estimated size is below spark.sql.autoBroadcastJoinThreshold, default 10485760 (10 MB).

2. Bucket both tables the same way. If both were written with bucketBy on the join key with the same bucket count, matching rows are already co-located in corresponding files, and the metastore records that fact.

(df.write.bucketBy(64, "rider_id").sortBy("rider_id")
   .saveAsTable("lakehouse.trips_bucketed"))

This is the only shuffle-free option for a genuinely large-to-large join.

3. Reuse an existing partitioning. If a DataFrame is already hash-partitioned on the join key, from a previous repartition(col) or an earlier shuffle on the same key, Catalyst can satisfy the join’s requirement without a new exchange. With RDDs this is explicit co-partitioning plus cache.

flowchart TB
  A["shuffle needed?"] --> B{"one side small?"}
  B -->|"yes"| C["broadcast: no shuffle"]
  B -->|"no"| D{"both bucketed<br/>same key, same count?"}
  D -->|"yes"| E["bucketed join: no shuffle"]
  D -->|"no"| F{"already partitioned<br/>on the key?"}
  F -->|"yes"| G["reuse: no shuffle"]
  F -->|"no"| H["sort-merge join: shuffle"]

The reduce side pulling its blocks from every map output. It is the half that crosses the network, and the knobs govern how aggressively it does so.

Raising the in-flight limits increases throughput and memory pressure; lowering them is gentler on a loaded cluster.

But the important answer is that fetch-side tuning is rarely the fix. A FetchFailedException almost always means the executor holding those blocks is gone, was killed for exceeding memory, or is too overwhelmed to serve. Raising retries makes the job take longer to fail rather than succeed.

So the diagnostic order is:

1. Find the executor that was serving those blocks and why it died. The executors tab shows loss reasons.
2. If it was killed for memory, fix that: overhead, skew, or partition size.
3. Consider an external shuffle service, so shuffle output survives the executor.
4. Only then touch fetch settings, and only with evidence of genuinely slow but healthy transfers.

The reduce side could not get shuffle blocks it needs. The block was written, but whatever was serving it is no longer able to.

The causes, most common first:

1. The executor died. Killed by the cluster manager for exceeding memory limits, or crashed with an OutOfMemoryError. Its shuffle files went with it.
2. The executor is alive but overwhelmed, so fetch requests time out while it is busy with GC or computation.
3. The node died, taking its local disk with it.
4. Disk full, so the map side could not write or the service cannot read.
5. Genuine network problems, which are the rarest in practice.

flowchart LR
  A["FetchFailedException"] --> B["look at the EXECUTORS tab,<br/>not the network config"]
  B --> C{"why did it die?"}
  C -->|"killed for memory"| D["overhead, skew,<br/>or partition size"]
  C -->|"OOM"| E["partition too large"]
  C -->|"still alive"| F["GC pressure or overload"]

What Spark does automatically: DAGScheduler treats it as lost map output and resubmits the upstream stage to regenerate it. So the job often recovers, just slowly, which is why you may see a stage run twice.

What to do: fix the executor deaths. An external shuffle service also helps directly, because shuffle blocks are then served by a node-level daemon that outlives the executor.

Spill is data written to disk because it did not fit in execution memory. The stage page reports it as two numbers: spill (memory), the in-memory size of what was spilled, and spill (disk), the serialized size actually written.

It happens on either side: the map side while sorting and buffering output, or the reduce side while merging fetched blocks and aggregating.

What it means: the per-task working set exceeds the execution memory each task gets. That memory is roughly the unified pool divided by the number of concurrently running tasks on that executor.

executor memory 16g
  - 300 MB reserved
  x spark.memory.fraction 0.6          -> ~9.6 GB unified pool
  / spark.executor.cores 4             -> ~2.4 GB per running task

So the fixes, in order:

1. More partitions, so each task’s share is smaller. This is usually the right answer and costs nothing but a config.
2. Fewer cores per executor, which gives each running task a larger slice of the same pool.
3. Check for skew, because one oversized partition spills while its peers do not.
4. More memory, last, because it treats the symptom.

A little spill is not an emergency; Spark is designed to spill rather than fail. Heavy spill relative to shuffle size means you are paying disk I/O for work that should have been in memory.

Because coalesce is a narrow dependency, so it creates no stage boundary, and the reduced parallelism propagates backwards into the stage that feeds it.

flowchart TB
  subgraph C["coalesce(1): ONE stage, 1 task total"]
    R1["read 200 partitions"] --> M1["expensive transform"] --> Z1["coalesce(1)"] --> W1["write 1 file"]
  end
  subgraph R["repartition(1): TWO stages"]
    R2["read 200 partitions"] --> M2["expensive transform<br/><b>200 tasks</b>"] --> S2[("shuffle")] --> W2["write 1 file<br/>1 task"]
  end

With coalesce(1), the read and the expensive transform are in the same stage as the single-partition write, so the whole pipeline runs in one task on one core. A job that should take minutes takes hours, and the cluster sits idle.

repartition(1) inserts a shuffle, which costs a full redistribution but keeps the upstream stage at full parallelism.

So the rule:

A middle path is often best: coalesce(200) to trim file count without a shuffle, rather than collapsing to one. And if you need exactly one file, ask whether you actually do, because a single-file requirement usually comes from a downstream tool that could read a directory instead.

UnifiedMemoryManager does the arithmetic in a fixed order:

1. Subtract 300 MB of reserved memory, a constant for Spark’s own overhead.
2. Take spark.memory.fraction of what remains, default 0.6, as the unified pool shared by execution and storage.
3. What is left, 40% of the post-reserved heap, is user memory: your own data structures, UDF state, and Spark internal metadata.

--executor-memory 16g  (16,384 MB)
  - 300 MB reserved              -> 16,084 MB usable
  x 0.6 memory.fraction          -> 9,650 MB unified pool
      of which storageFraction 0.5 -> 4,825 MB storage floor
  remaining 40%                  -> 6,434 MB user memory
  + memoryOverhead               -> ~18,022 MB requested from the cluster

flowchart TB
  C["container: heap + overhead"] --> H["heap 16g"]
  C --> O["overhead, off-heap and native"]
  H --> R["300 MB reserved"]
  H --> U["unified pool, 60%<br/>execution + storage"]
  H --> M["user memory, 40%"]
  U --> E["execution: shuffles, joins, sorts"]
  U --> S["storage: cached blocks, broadcasts"]

Two things people miss: the pool is shared between execution and storage combined, which is usually less than expected; and the container asks the cluster manager for heap plus spark.executor.memoryOverhead, which is why a 16g executor consumes about 18g of cluster capacity.

Off-heap is separate and off by default: spark.memory.offHeap.enabled is false.

They are two uses of one pool.

They borrow from each other, which is what “unified” means. If nothing is cached, execution can use the whole pool. If no shuffle is running, cached blocks can fill it.

spark.memory.storageFraction is not a partition; it is the floor below which cached blocks will not be evicted. Cached data can grow past it when execution is idle, but cannot be pushed below it.

Before Spark 1.6 these were static pools, and the tuning pain of that is why unification happened: a job that cached nothing still had memory reserved for caching it could not use.

The consequence for tuning: spark.memory.fraction is rarely worth changing, because raising it steals from user memory, which UDFs and your own objects need. If you are tempted to raise it, the real problem is usually partition size.

No, and the asymmetry is the whole answer.

Execution can evict cached blocks, down to the storage floor. Storage can never evict execution.

flowchart LR
  E["execution needs memory"] -->|"evicts cached blocks<br/>down to storageFraction"| S["storage"]
  S -.->|"never evicts"| E

The reason is correctness and progress: a task that cannot get execution memory can only spill or fail, whereas a cached block that is evicted can simply be recomputed from lineage. So Spark prioritises the thing that cannot recover.

The practical consequences, and these are the interview-worthy part:

• Caching a large DataFrame cannot fail a shuffle. Execution will take the memory it needs.
• But it can make a shuffle spill, because the cache occupied the pool first and execution had to reclaim it, and because whatever stays above the floor is unavailable.
• So adding a cache() can make a job slower, which is a counterintuitive result worth being able to explain.
• And an under-populated cache is the worst outcome: you paid to write it, execution evicted most of it, and now every use partly recomputes.

Check the storage tab. A cached dataset showing 40% in memory is costing you without paying you back, and either needs MEMORY_AND_DISK, a serialized level, or not to be cached at all.

On-heap memory is allocated inside the JVM heap and managed by the garbage collector. Off-heap is allocated outside it, through sun.misc.Unsafe, and is not subject to GC.

Why off-heap exists: a large heap full of small objects makes full garbage collections slow, and GC pauses stall the executor. Storing data off-heap in a compact binary form removes it from the collector’s view entirely.

This is what Tungsten is built on. Spark’s binary row format stores rows as contiguous bytes with a known layout, so it can be read field by field without constructing JVM objects, and can live off-heap.

--conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=4g

When to bother: very large executors where GC time is a visible fraction of task time, or heavy shuffle and aggregation workloads. Note that off-heap size counts toward the container total, so it must fit inside what the cluster manager granted, which usually means accounting for it in memoryOverhead.

Memory requested from the cluster manager in addition to the JVM heap, to cover everything that is not heap:

• Off-heap allocations and Tungsten buffers.
• The Python worker process, for PySpark.
• Native libraries, and the JVM’s own non-heap regions such as metaspace, thread stacks and code cache.
• Netty’s direct buffers for shuffle transfer.

spark.executor.memoryOverhead defaults to a fraction of executor memory with a minimum floor, so a 16g executor requests roughly 18g in total.

--conf spark.executor.memory=16g --conf spark.executor.memoryOverhead=3g

Why it is the answer to a specific failure. When the cluster manager kills a container for exceeding its memory limit, that is the total being exceeded, not the heap. A JVM OutOfMemoryError means the heap; a container kill means the total. They have different fixes, and telling them apart is the skill:

PySpark jobs need noticeably more overhead than JVM ones, because every executor runs Python workers whose memory is entirely outside the heap. A PySpark job being killed while the heap looks healthy is the classic case.

Something is pulling data into the driver, or the driver is tracking too much metadata.

The causes, most common first:

1. collect() or toPandas() on a large result. This is by far the most common.
2. A broadcast that is not small. The driver must assemble the whole table before sending it, so broadcast() on a multi-gigabyte side fails on the driver.
3. A huge number of tasks. The driver tracks metrics and state per task, so a job with hundreds of thousands of tasks needs real memory just for bookkeeping.
4. A deep or wide query plan. Analysis and optimisation hold the plan tree in driver memory; thousands of chained withColumn calls or very wide schemas add up.
5. Accumulating results in a loop, appending to a driver-side list across iterations.

spark.driver.memory defaults to 1g, which is a coordinator’s budget.

The distinction from spark.driver.maxResultSize, default 1g, is worth making: that config aborts the job with a clear named error before the heap is exhausted. So a maxResultSize error is Spark protecting you, and an OutOfMemoryError is what happens when the data arrived some other way, such as through a broadcast.

The fix is usually structural, not a larger driver: write instead of collect, aggregate instead of transfer, and check the plan for an accidental broadcast.

A single task needing more heap than it can get. The usual sources:

1. A partition too large. The task must hold its working set, and if that exceeds its share of the unified pool plus what can spill, it fails.
2. Skew. One key concentrating rows into one task, so that task alone is oversized.
3. A high-cardinality aggregation, where the hash map of groups grows large.
4. groupByKey on a hot key, because all values for one key must fit on the reduce side.
5. A UDF materialising too much, for example list(rows) inside mapPartitions, or an applyInPandas group that does not fit.
6. Too many cores per executor, so the pool is divided too many ways.

The arithmetic that explains most cases:

per-task execution memory ~= unified pool / concurrent tasks on the executor
                          ~= 9.6 GB / 4 cores
                          ~= 2.4 GB

So the fixes in order:

1. More partitions. Halving partition size halves the working set, and this is usually the correct answer.
2. Check and fix skew, since one bad partition is invisible in an average.
3. Fewer cores per executor, which gives each task a bigger slice of the same pool.
4. A serialized storage level if caching is competing.
5. More executor memory, last.

Raising memory first often just moves the failure to a larger dataset next month.

When both conditions hold: the dataset is used more than once, and producing it was expensive.

base = (spark.read.parquet(path)
    .filter(F.col("started_at") >= "2026-09-01")
    .join(dims, "city_id"))        # expensive: a scan and a shuffle

base.cache()
base.count()                        # populates

by_city = base.groupBy("city_id").count()
by_rider = base.groupBy("rider_id").count()     # reuses the cache

Without the cache, base is recomputed for each aggregation, including the scan and the join.

Good candidates:

• A cleaned, filtered base dataset feeding several outputs.
• An iterative algorithm reading the same data each pass.
• A dataset used by both branches of a self-join or a union.
• Anything you will query interactively in a notebook.

And cache deliberately, then verify. Call an action to populate it, and check the storage tab: if it is not fully in memory, the cache may be costing more than it saves. Release it when done:

base.unpersist()

One subtlety: caching is by plan, not by variable. Two DataFrames with identical plans share the cache entry, which is occasionally surprising and occasionally very useful.

Four cases where it costs more than it saves.

1. Used once. You pay to write the cache and gain nothing. This is the most common mistake, and in a notebook it is everywhere.
2. Cheap to recompute. A plain scan of a small Parquet file is faster to redo than to serialise into memory.
3. Memory already tight. The cache occupies the unified pool, execution evicts it, and you end up partly recomputing anyway while having made every shuffle more likely to spill.
4. Immediately before one big shuffle. The cache and the shuffle want the same pool, and the shuffle is the thing that cannot recover.

There is also a correctness-adjacent case: caching a non-deterministic DataFrame pins one particular result, which may be what you want or may hide a bug. A rand() or a current_timestamp() in the plan behaves differently cached than uncached.

The anti-pattern to name explicitly:

# caching everything, which is caching nothing usefully
df1.cache(); df2.cache(); df3.cache(); df4.cache()

Each competes for the same pool, most get evicted, and every stage now spills more. Cache the one dataset that is genuinely reused, populate it, verify it in the storage tab, and unpersist it when the reuse is over.

Four, and the choice is a memory-versus-CPU trade.

from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

The serialized variants are the useful lever. A deserialized cache stores rows close to their in-memory representation, which is fast to read and large. A serialized cache is far more compact, so more of the dataset stays in memory, at the cost of CPU to decode on every read. When the storage tab shows a cache only partly resident, serialized is usually the better trade.

The _2 replicated variants put a copy on a second node so an executor loss does not force recomputation. They double the memory cost, so they are for the case where recomputation is genuinely very expensive.

Note DataFrame caching already stores data in a compressed columnar form internally, controlled by spark.sql.inMemoryColumnarStorage.compressed, so the gap between deserialized and serialized is smaller than for RDDs.

df.unpersist()                    # one dataset, blocking=False by default
spark.catalog.clearCache()        # everything
spark.sql("UNCACHE TABLE trips")  # a named table

It matters because cached blocks hold storage memory for the lifetime of the application, not the lifetime of the variable. Python’s garbage collector reclaiming a DataFrame object does not release the blocks on the executors: Spark’s block manager still holds them.

So in a long-running application, a notebook, a streaming job, or a multi-stage pipeline, forgotten caches accumulate. Every later stage then has less execution memory, spills more, and runs slower, with no obvious cause. The symptom is a pipeline that gets slower toward the end for no reason visible in the plan.

The habit worth building:

base.cache()
base.count()
# ... the reuse that justified the cache ...
base.unpersist()

Check what is actually held in the storage tab of the UI, which lists every cached dataset with its size and memory fraction. Anything there that you no longer read is pure cost.

One note on unpersist(blocking=True): it waits for the blocks to actually be removed, which is worth using when you need the memory freed before the next stage rather than eventually.

No. cache() and persist() are lazy: they mark the dataset as one to store, and the blocks are written during the next action that computes it.

df.cache()          # nothing happens yet
df.count()          # computes, and populates the cache on the way
df.groupBy(...)     # now reads from the cache

This matters in two ways:

• The first action pays the cost. It computes the dataset and writes the cache, so it is slower than the uncached version. The saving starts with the second use.
• If you never call an action, nothing is cached, which is a silent no-op people hit when they cache and then only inspect the schema.

The idiom is to force population deliberately with a cheap action, usually count(), so the cost lands where you expect rather than inside the first real query.

The exception is SQL. CACHE TABLE is eager by default:

CACHE TABLE trips;          -- populates immediately
CACHE LAZY TABLE trips;     -- behaves like df.cache()

That asymmetry between the DataFrame API and SQL is a favourite detail to probe, because it is easy to assume they behave the same.

Checkpointing writes a dataset to reliable storage and truncates its lineage. Caching stores a copy while keeping the lineage intact.

spark.sparkContext.setCheckpointDir("hdfs:///checkpoints")
df = df.checkpoint()          # eager by default for DataFrames

The reason lineage truncation matters: with a long chain of transformations, a single lost partition means replaying the whole chain, and the plan itself grows until analysis is slow or overflows the stack. Checkpointing cuts the chain, so recovery reads a file instead of recomputing a hundred steps.

Where you need it: iterative algorithms that build lineage per iteration, and Structured Streaming, where checkpointing is mandatory and stores offsets and state.

localCheckpoint() truncates lineage using executor storage instead of reliable storage: much faster, but lost if the executor is lost, so it trades durability for speed.

A common pattern is to do both, cache so the write is not recomputed, then checkpoint.

Because lineage is a data structure held on the driver, and it grows whether or not the data does.

Three separate problems:

1. Planning cost. Every action re-analyses and re-optimises the plan. Catalyst walks the tree repeatedly, so a tree with thousands of nodes makes planning a measurable fraction of the job. In a loop, planning time grows each iteration.
2. Stack depth. Many Catalyst rules are recursive over the tree. A deep enough plan overflows the thread stack, producing a java.lang.StackOverflowError that looks nothing like a data problem.
3. Recovery cost. One lost partition replays the entire chain that produced it.

The classic trigger is a loop:

for i in range(200):
    df = df.withColumn(f"c{i}", F.col("x") + i)      # 200 levels deep

or an iterative algorithm where each pass depends on the last.

The remedies:

• Checkpoint periodically, every few iterations, which truncates the chain.
• Collapse the operations. One select with 200 expressions instead of 200 withColumn calls.
• Raise -Xss on the side that threw, as a stopgap for the stack overflow specifically.

The tell that distinguishes this from other slowness: the job spends its time before any task starts. If the stages are quick but the gap between them is long, you are looking at planning, not execution.

spark.serializer defaults to org.apache.spark.serializer.JavaSerializer.

Java serialization is general and slow, and produces large output. Kryo is substantially more compact and faster:

--conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.kryo.registrationRequired=true

conf.registerKryoClasses(Array(classOf[Trip], classOf[City]))

Registering your classes matters: unregistered classes are written with their full class name per record, which throws away much of the benefit. registrationRequired=true turns a missed registration into an error rather than silent bloat.

Where it does and does not help is the important half:

DataFrames mostly use Tungsten encoders, which are generated per schema and already compact, so they bypass spark.serializer for row data. That is why “set Kryo” is standard advice for RDD-era code and much less impactful for modern DataFrame pipelines.

So: set it if you have RDD-heavy code or serialized caching. If your job is pure DataFrame, expect little and look elsewhere.

Spark’s execution backend, aimed at making the JVM behave more like hand-written memory-managing code. Three parts:

1. Binary row format. UnsafeRow stores a row as a contiguous block of bytes with a fixed-size region for primitives and offsets into a variable-length region. Fields are read by offset arithmetic, with no objects constructed.
2. Explicit memory management. Spark allocates and tracks memory itself, on or off heap, rather than relying on the garbage collector for data.
3. Whole-stage code generation. Operators in a stage are fused into one generated Java method, so there is no per-row virtual call chain.

Why it matters, concretely: a JVM object per row costs a header, pointer indirection, and GC attention. A million-row partition becomes a million objects. UnsafeRow makes it one byte array plus offsets.

JVM object row   : header + pointers + boxed fields, GC-visible
UnsafeRow        : contiguous bytes, read by offset, GC-invisible

This is the technical answer to “why do DataFrames beat RDD code”. It is not that the DataFrame API is better written; it is that RDDs must materialise your objects because Spark does not know their shape, while a DataFrame’s schema lets Tungsten use a compact layout and generate code against it.

It also explains why a Python UDF is expensive: it forces rows out of that binary format, across a process boundary, and back.

Read the GC Time column in the executors tab of the UI, and compare it to task time. A rough guide: consistently above 10% of task time is worth investigating, and above 20% is a problem.

The causes, and the fix for each:

To see what the collector is actually doing, log it:

--conf "spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps"

Two structural points worth making:

• Off-heap storage removes data from the collector’s view entirely, which is the real fix for GC-bound shuffle and aggregation work.
• Very large heaps are not free. A 64g executor can spend a long time in a full collection, which is one reason the usual advice is several medium executors rather than one enormous one.

And distinguish GC pressure from spill: GC is CPU spent on memory management, spill is disk I/O from insufficient memory. They often appear together, but the remedies differ.

For every cached dataset: its name, storage level, the number of cached partitions, the fraction held in memory, and the sizes in memory and on disk.

The single most useful number is the memory fraction. What it tells you:

That third row is the trap: an RDD cached MEMORY_ONLY that only half fits silently recomputes half the partitions on every use, which looks like the cache having no effect at all.

What to do with the information:

• Partial residency means either switch to a serialized level so more fits, accept MEMORY_AND_DISK, cache a smaller projection, or do not cache.
• Several datasets listed when you only meant to cache one is the “cache everything” anti-pattern, and they are competing for the same pool.
• An empty tab on a job you expected to use a cache means the cache() call never populated.

It is also the place to confirm an unpersist() actually released the blocks rather than just dropping your reference.

Five, and Catalyst picks one per join based on size estimates, join type and hints.

The selection order is roughly: can it broadcast, then can it shuffle-hash, then sort-merge, and for non-equi conditions only the nested-loop options are available at all.

flowchart TB
  A["equi-join?"] -->|"no"| N["BroadcastNestedLoop<br/>or Cartesian"]
  A -->|"yes"| B{"one side <<br/>autoBroadcastJoinThreshold?"}
  B -->|"yes"| BH["BroadcastHashJoin"]
  B -->|"no"| C{"one side small enough<br/>to hash per partition?"}
  C -->|"yes"| SH["ShuffleHashJoin"]
  C -->|"no"| SM["SortMergeJoin"]

The last two rows of the table are the ones to notice in a plan: seeing CartesianProduct or BroadcastNestedLoopJoin where you expected an equi-join almost always means the join condition is not what you think.

Spark sends the entire smaller side to every executor, where each builds a hash table in memory. Each partition of the large side then joins locally, so the large side is never shuffled.

flowchart TB
  D["driver collects<br/>the small side"] --> E1["executor 1<br/>hash table"]
  D --> E2["executor 2<br/>hash table"]
  D --> E3["executor 3<br/>hash table"]
  L1["large partition"] --> E1
  L2["large partition"] --> E2
  L3["large partition"] --> E3

Spark chooses it automatically when the estimated size of one side is below spark.sql.autoBroadcastJoinThreshold, default 10485760, which is 10 MB. Set it to -1 to disable automatic broadcasting.

spark.conf.get("spark.sql.autoBroadcastJoinThreshold")   # 10485760

Why it is the fastest option: no shuffle at all on the large side, which removes the network transfer, the disk write and the stage barrier.

The costs, which a good answer names:

• The driver assembles it first. The small side is collected to the driver before being broadcast, so a too-large broadcast fails on the driver, not the executors.
• Memory per executor. Every executor holds a full copy, so 200 MB broadcast across 50 executors is 10 GB of cluster memory.
• It is bounded by join type. A full outer join cannot be broadcast, and for a left outer join only the right side can be.

from pyspark.sql.functions import broadcast
joined = trips.join(broadcast(cities), "city_id")

SELECT /*+ BROADCAST(c) */ *
FROM trips t JOIN cities c ON t.city_id = c.city_id

The available hints, and what each forces:

To prevent broadcasting globally:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

When forcing is legitimate: Spark’s estimate is wrong, usually because statistics are missing or the side is derived through several transformations, and you know it is small.

When it backfires: hinting BROADCAST on something genuinely large. The driver must collect it first, so you move the failure from a slow join to a driver OutOfMemoryError, which is harder to diagnose. A hint overrides the safety check, so it is a claim you are making about the data.

A better habit than hinting is often to fix the estimate: ANALYZE TABLE ... COMPUTE STATISTICS, or filter and project the small side first so it is genuinely below the threshold. And with AQE enabled, a runtime switch to a broadcast often happens without a hint at all.

How long Spark will wait for a broadcast to be built and distributed before failing the query. The default is 300 seconds.

The error is explicit about the config, which is a clue people sometimes miss: it tells you to either raise the timeout or disable broadcast joins.

spark.conf.get("spark.sql.broadcastTimeout")   # 300

Why raising it is usually the wrong move. Hitting a five-minute timeout to build what should be a small hash table means the side is not small. The sequence of events is: Spark estimated it as broadcastable, began collecting it to the driver, and that collection is taking minutes.

So the diagnosis order:

1. Check the actual size of the side being broadcast. If it is hundreds of megabytes, the estimate was wrong.
2. Check whether a hint forced it. A BROADCAST hint overrides the size check entirely.
3. Check for a missing filter. A side that should be small after filtering is being broadcast before the filter, or the filter cannot push down.
4. Then consider the timeout, legitimately, if the table is genuinely small but the cluster is slow or contended.

The alternative to raising it is to remove the broadcast and let the join shuffle, which is slower per-query but predictable:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

Because its memory requirement does not scale with the size of either side.

A hash join must hold one side’s hash table in memory per partition. A sort-merge join sorts both sides, then walks them together with two cursors, so it only needs buffers for the merge, not the whole partition.

left  (sorted):  a a b c d ...
right (sorted):  a b b d e ...
                 ^ ^            advance the smaller cursor, emit matches

And the sort can spill to disk, so a partition larger than memory still completes rather than failing.

spark.sql.join.preferSortMergeJoin defaults to true, which biases the planner toward it over shuffle hash join.

So the trade is CPU for robustness: you pay the sort in exchange for bounded memory. For two large tables that is the right trade, which is why it is the default.

A useful corollary: if both sides are already sorted on the join key, for example from bucketing with sortBy, the sort is skipped and sort-merge becomes very cheap.

Both sides are shuffled on the join key, then for each partition Spark builds a hash table from the smaller side and probes it with the larger.

It beats sort-merge when one side is small enough to hash per partition but too large to broadcast, because it skips the sort entirely.

sort-merge   : shuffle both, sort both, merge
shuffle hash : shuffle both, hash the small side, probe

The conditions Spark requires before choosing it:

• The join is an equi-join.
• One side is estimated small enough that a per-partition hash table fits, judged against spark.sql.autoBroadcastJoinThreshold multiplied by the number of shuffle partitions.
• spark.sql.join.preferSortMergeJoin is false, or the cost model prefers it.

Because preferSortMergeJoin defaults to true, Spark picks it less often than you might expect, and the usual reason is exactly that conservatism: sort-merge cannot run out of memory, and shuffle hash can.

To force it when you know the shape:

SELECT /*+ SHUFFLE_HASH(c) */ * FROM trips t JOIN cities c ON t.city_id = c.city_id

The honest summary for an interview: it occupies a narrow band between broadcast and sort-merge, and in modern Spark, AQE converting a planned sort-merge into a broadcast at runtime covers much of that band more safely.

Both avoid a shuffle, for entirely different reasons and at different scales.

A broadcast join is a runtime decision about a small side. A bucketed join is a storage layout decision that pays off on every future join.

(trips.write.bucketBy(64, "rider_id").sortBy("rider_id")
   .mode("overwrite").saveAsTable("lakehouse.trips_b"))
(riders.write.bucketBy(64, "rider_id").sortBy("rider_id")
   .mode("overwrite").saveAsTable("lakehouse.riders_b"))

spark.table("lakehouse.trips_b").join(spark.table("lakehouse.riders_b"), "rider_id")

The conditions are strict, and naming them is the good answer: both sides bucketed, on the same column, with the same number of buckets. A mismatch in bucket count means Spark shuffles anyway, silently, and you have paid the write cost for nothing.

So bucketing is the only shuffle-free route for large-to-large joins, and its drawback is inflexibility: the bucket count is baked in, and a join on a different key gets no benefit.

Because Spark broadcasts on an estimate, and the estimate can be wrong or unavailable.

The reasons, most common first:

1. No statistics. For a file-based table Spark falls back to the total file size, which can be far larger than the in-memory size after filtering, or unavailable for some sources.
2. The side is derived. After a filter and a join, the estimate is a computed guess. Spark multiplies selectivity factors, and those compound badly.
3. The filter cannot push down, so the estimate is of the unfiltered table.
4. Compression. A 50 MB Parquet file may be 500 MB in memory, and Spark’s estimate accounts for that, so a “small” file can exceed the threshold.
5. The join type forbids it. A full outer join cannot be broadcast, and for outer joins only the nullable side can be.

The fixes, in order of preference:

ANALYZE TABLE lakehouse.cities COMPUTE STATISTICS FOR ALL COLUMNS;

cities_small = cities.filter(F.col("active")).select("city_id", "name")
trips.join(F.broadcast(cities_small), "city_id")

And the modern answer: adaptive query execution largely solves this. Because AQE re-plans after the shuffle stage completes, it knows the actual size and can convert a planned sort-merge join into a broadcast. So on Spark 3.2 and later, a missing broadcast at plan time often becomes a broadcast at runtime, which is why the SQL tab shows a different plan than explain() did.

Cost-based optimisation uses table and column statistics to make planning decisions that rule-based optimisation cannot: which join order is cheapest, and which physical strategy to pick.

spark.sql.cbo.enabled defaults to false.

Even when enabled it needs statistics to exist:

ANALYZE TABLE lakehouse.trips COMPUTE STATISTICS FOR ALL COLUMNS;

spark.conf.set("spark.sql.cbo.enabled", "true")
spark.conf.set("spark.sql.cbo.joinReorder.enabled", "true")

What it can do that rules cannot: reorder a multi-way join so the most selective join happens first, which changes the size of every intermediate result.

Why it is off by default, and why that is defensible: CBO depends on statistics being present, current and representative. In a lakehouse where tables are written continuously, statistics go stale quickly, and a confidently wrong estimate is worse than no estimate. Collecting column statistics on a large table is also itself an expensive scan.

And adaptive execution has taken over much of its job using real runtime numbers rather than stored estimates, which do not go stale. So the modern answer to “should I enable CBO” is usually: make sure AQE is on, and reach for CBO only for complex multi-way joins over stable tables where join order is the problem.

After a shuffle stage completes, AQE knows the actual size of each side, so it can replace the planned strategy with a better one. The main conversion is demoting a sort-merge join to a broadcast hash join.

flowchart LR
  A["plan time:<br/>estimate says 800 MB<br/>-> SortMergeJoin"] --> B["stage runs"]
  B --> C["actual: 6 MB"]
  C --> D["re-plan:<br/>BroadcastHashJoin"]

This is the single most valuable thing AQE does, and the reason is structural: planning-time estimates on derived data are usually wrong. Spark can estimate a base table it has statistics for, but after two filters and a join the number is a product of guessed selectivities. After the shuffle, it is a measurement.

The relevant setting is spark.sql.adaptive.autoBroadcastJoinThreshold, which if unset falls back to the static spark.sql.autoBroadcastJoinThreshold.

Two consequences worth knowing:

• explain() lies about the final plan. Before execution the tree is AdaptiveSparkPlan isFinalPlan=false, so what you read is the planned shape. The SQL tab after the query shows what ran.
• There must be a shuffle for AQE to act at. A query with no exchange gets no adaptive help, and the first stage is always planned blind.

A related optimisation follows automatically: once the join becomes a broadcast, the already-written shuffle can be read locally rather than over the network.

An AQE optimisation that avoids network traffic when a join strategy changes at runtime.

The situation: Spark planned a sort-merge join, so both sides were shuffled. Then AQE discovered one side is small and converted the join to a broadcast. The shuffle files already exist, and with a broadcast join each task no longer needs a specific partition from every map output; it just needs the rows. So instead of fetching them over the network in the shuffle pattern, each executor reads the shuffle blocks it wrote locally.

spark.sql.adaptive.localShuffleReader.enabled defaults to true.

without: every reduce task fetches from every map output    (network)
with   : each executor reads its own local shuffle blocks   (local disk)

In the plan it appears as LocalShuffleReader above the exchange.

The reason to know it exists is mainly for reading plans: seeing LocalShuffleReader tells you AQE changed a join strategy after the shuffle had already been written, which is a useful signal that your plan-time estimates were wrong. That in turn tells you statistics might be worth collecting, or that a filter should be pushed earlier so the plan is right the first time.

Compare the planned tree with what actually ran, because they are different objects.

Before execution, explain() shows the planned shape wrapped in an adaptive node:

AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [city_id#21], [city_id#40], Inner
   :- Sort [city_id#21 ASC], false, 0
   :  +- Exchange hashpartitioning(city_id#21, 200)
   ...

After execution, the same query’s entry in the SQL tab of the UI shows isFinalPlan=true and the strategies AQE settled on. You can also re-run explain() on an already-executed DataFrame and get the final plan.

What to look for as evidence AQE acted:

The practical habit: never conclude anything about a modern Spark query from explain() alone. It tells you the starting point. The SQL tab tells you the result, and when the two differ, the difference is exactly what AQE contributed.

Both filter the left side by whether a match exists on the right, and neither returns any right-side columns.

trips.join(cities, "city_id", "left_semi")    # rows WITH a match
trips.join(cities, "city_id", "left_anti")    # rows WITHOUT a match

Why they are better than the alternatives, which is the point of the question:

• Against an inner join plus select: a semi join stops at the first match, so a left row matching 100 right rows is emitted once. An inner join emits it 100 times, and then you have to deduplicate.
• Against a left_outer plus isNull filter for the anti case: the outer join carries all the right-side columns through the shuffle before you discard them.

So semi and anti joins are both correct where the alternatives need a distinct, and cheaper, because less data crosses the shuffle.

The null behaviour is worth knowing: for a left anti join, a null key on the left never matches, so those rows are returned, which is usually what you want but occasionally surprising.

Null never equals null in SQL semantics, so a null key matches nothing. Three consequences and their handling:

1. Inner joins silently drop null keys. If the null means “not applicable”, that is correct. If it means “unknown but should match”, you must say so:

a.join(b, a.city_id.eqNullSafe(b.city_id))      # null == null is true here

2. Null keys are a skew source. Every null hashes to the same partition, so a column that is 30% null puts 30% of the rows in one task. If they carry no meaning, drop them before the join:

a.filter(F.col("city_id").isNotNull()).join(b, "city_id")

Or replace them with a distributed sentinel, which is salting for nulls:

a.withColumn("join_key", F.coalesce("city_id",
    F.concat(F.lit("__null_"), (F.rand() * 16).cast("int"))))

3. Outer joins produce nulls too, so after a left outer join you cannot distinguish “matched with a null value” from “did not match” by looking at the column. Add a marker before joining if you need to tell them apart:

b_marked = b.withColumn("_matched", F.lit(True))
a.join(b_marked, "city_id", "left").withColumn(
    "had_match", F.coalesce("_matched", F.lit(False)))

The habit: check the null count on every join key before writing the join. It answers both the correctness and the skew question at once.

A join condition Spark cannot treat as an equality, so it has no key to partition on and must compare every pair.

The usual causes:

1. A non-equi condition. ON a.ts BETWEEN b.start AND b.end has no equality, so there is nothing to hash.
2. The condition moved into a where after a cross join. Semantically equivalent SQL, but if Spark cannot push it into the join it evaluates the product first.
3. A join on an expression that Catalyst cannot match to a key, for example ON cast(a.id AS string) = b.id.
4. A missing condition entirely, such as a.join(b) with no key, or a multi-column join where one predicate was forgotten.

In the plan it shows as CartesianProduct or BroadcastNestedLoopJoin:

CartesianProduct
:- FileScan parquet [...]
+- FileScan parquet [...]

How to catch it: read the plan whenever a join is unexpectedly slow. A Cartesian on two 1-million-row tables is a trillion comparisons, so the symptom is a stage that never finishes rather than one that is merely slow.

There is a guard: spark.sql.crossJoin.enabled historically had to be set for implicit cross joins, forcing you to be explicit with crossJoin().

For a legitimate range join, the usual technique is to add an equality that buckets both sides, for example joining on a truncated date as well as the range condition, so Spark has a key to partition on and the nested-loop comparison happens only within each bucket.

Using the result of a filter on one side of a join to prune partitions of the other side at runtime, rather than scanning all of them.

The classic star-schema shape: a large partitioned fact table, joined to a small filtered dimension table.

SELECT t.*
FROM trips t JOIN cities c ON t.city_id = c.city_id
WHERE c.region = 'west'

Static partition pruning cannot help, because the filter is on c.region while trips is partitioned by city_id. Without DPP, Spark scans every partition of trips and discards most of them after the join.

With DPP, Spark runs the dimension side first, collects the matching city_id values, and injects them as a partition filter on the fact scan.

flowchart LR
  A["filter cities<br/>region = 'west'"] --> B["city_ids: sf, la, sea"]
  B --> C["inject as a partition filter"]
  C --> D["scan only those<br/>trips partitions"]

The conditions: the fact table must be partitioned on the join key, the join must be an equi-join, and the dimension side must be filtered and broadcastable.

In the plan it appears as a dynamicpruningexpression in the fact-side PartitionFilters.

The payoff is often the largest single win available on a star schema, because it changes how much data is read rather than how efficiently it is processed.

Broadcasting is off the table, so the work is in reducing what enters the shuffle and making the shuffle well-shaped.

In order:

1. Reduce both sides first. Filter and project before joining. This is the largest lever and the most often skipped:

t = trips.filter(F.col("started_at") >= "2026-09-01").select("trip_id", "rider_id", "fare_amount")
r = riders.filter(F.col("active")).select("rider_id", "segment")

Sometimes this alone makes one side broadcastable, which changes the problem entirely.

2. Check whether one side can become small. An aggregation before the join often shrinks it by orders of magnitude.

3. Bucket both tables on the join key with the same bucket count, if this join is a recurring pattern. That removes the shuffle permanently.

4. Size the shuffle. Aim for partitions in the low hundreds of megabytes, and set spark.sql.adaptive.advisoryPartitionSizeInBytes rather than guessing a count.

5. Handle skew on the join key, since a large-to-large join is where skew hurts most. AQE skew handling first, then salting.

6. Avoid re-shuffling. If you then aggregate on the same key, one exchange serves both. If you aggregate on a different key, you pay a second shuffle, so consider whether the order can change.

The one thing not to do is hint BROADCAST and hope.

A shuffle. Every Exchange in a physical plan is one redistribution of data, and therefore one stage boundary.

Its description carries the useful detail:

Exchange hashpartitioning(city_id#21, 200), ENSURE_REQUIREMENTS, [plan_id=141]
Exchange rangepartitioning(started_at#30 ASC NULLS FIRST, 200), ...
Exchange SinglePartition, ...

The ENSURE_REQUIREMENTS tag means the exchange was inserted because an operator above it required that partitioning, which is the normal reason.

How to use it:

• Count them to count your shuffles, and therefore your stages.
• Read the key to see whether consecutive operations shuffle on the same column, in which case one exchange can serve both.
• Watch for SinglePartition, which usually comes from a global orderBy, a coalesce(1), or a window with no partitionBy. That last one is a common accidental performance disaster: a window function without partitionBy sends the entire dataset through one task.

A shuffle whose output is read by more than one branch of the plan, rather than being computed twice.

It appears when the same subtree occurs in two places, for example in a self-join, a union of two aggregations over the same base, or a query referencing the same CTE twice.

:- Exchange hashpartitioning(city_id#21, 200)
:  +- HashAggregate(...)
+- ReusedExchange [city_id#40], Exchange hashpartitioning(city_id#21, 200)

Seeing it is good news: Catalyst recognised the common subtree and is reusing the shuffle files rather than recomputing the stage. In the UI this is also what produces “skipped stages”, shown in grey, which are not a problem but a saving.

The interesting case is its absence. If you expected reuse and do not see it, the two branches differ in some way you did not intend:

• A different filter, even a logically equivalent one written differently.
• A different column projection, so the plans are not identical.
• A non-deterministic expression such as rand() or current_timestamp(), which makes the subtrees non-equivalent by definition.

When reuse does not happen and you want it, cache() on the shared base is the explicit way to force one computation, at the cost of storage memory. Checking for ReusedExchange first tells you whether the cache is even necessary.

Six techniques, roughly in order of payoff.

1. Aggregate and join on the same key. If both operations need the same partitioning, one exchange satisfies both. Reordering a pipeline so the keys line up can halve the shuffles.

2. Repartition once, deliberately. If several operators key on the same column, one repartition(col) up front means each of them finds the partitioning already satisfied.

3. Replace distinct plus join with a semi join. distinct is a shuffle; a semi join achieves the same filtering in the join’s own shuffle.

trips.join(active_riders, "rider_id", "left_semi")     # not distinct + inner join

4. Broadcast what can be broadcast. A broadcast join has no exchange at all, so reducing a dimension until it fits under the threshold removes a shuffle outright.

5. Avoid unnecessary global ordering. orderBy costs a range-partitioning shuffle plus a sampling job. If you only need order within files, use sortWithinPartitions.

6. Bucket recurring join keys, so the shuffle is paid once at write time rather than on every read.

Then verify, because this is all measurable:

df.explain("formatted")     # count the Exchange nodes before and after

The mindset worth conveying: do not think of shuffles as something to tune, think of them as something to count, and then ask whether each one is earning its place.

Find the bottleneck before changing anything. The order matters more than any individual technique.

1. Open the SQL tab and find the slowest operator. The per-node timing tells you where the work is, which is often not where you assumed.
2. Open the stage page for that stage and read the task duration distribution: min, median, max.
3. Classify it. Almost every slow stage is one of four things:

1. Check the plan for structural mistakes before tuning: a missing broadcast, an accidental Cartesian, a filter that did not push down, columns you did not need.
2. Change one thing, re-run, and compare the same stage.

The answer that impresses is the negative one: say what you would not do first. You would not raise executor memory, not enable speculation, and not change spark.sql.shuffle.partitions until you knew which of the four it was. Tuning without a diagnosis is how jobs acquire a page of configs that nobody can justify.

How to use them together, which is the real skill:

• Start in SQL for a DataFrame job, because it maps operators to time. Starting in Stages tells you where but not what.
• Stages gives you the distribution. The summary metrics table with percentiles is the single most valuable view in the UI, because it separates skew from uniform slowness.
• Executors answers “why did it fail”. Loss reasons live here: killed for memory, heartbeat timeout, or exited.
• Environment settles arguments. If someone insists a config is set, this is the authoritative answer.
• Storage is where caches are exposed as a cost rather than assumed to be a benefit.

And remember the UI dies with the application, which is why event logging plus a history server is close to mandatory: most investigations happen after the job.

Skew, in almost every case. Confirm it and then fix the distribution.

Confirm: on the stage page, compare the max and median for both duration and shuffle read size. The second column is what distinguishes the two possible causes:

If it is the second, look at the executors tab for GC time and at which host the slow tasks ran on; a single bad machine shows up as all slow tasks on one executor.

If it is skew, fix it in this order:

1. Check whether AQE skew handling could fire. It needs a partition over 5x the median and over 256 MB, so on a smaller job it silently never triggers.
2. Broadcast the small side if there is one.
3. Salt the hot key.
4. Split the hot keys out and union.

And do not enable speculation. The duplicate task has the same rows and is equally slow, so it consumes a second slot for no benefit and masks the real cause by making it look like flaky infrastructure.

Spill means the per-task working set exceeded the execution memory that task could get. So reduce the working set, or increase the share.

The arithmetic that frames the fix:

per-task execution memory ~= unified pool / concurrent tasks per executor
                          ~= (heap - 300MB) x 0.6 / spark.executor.cores

Changes in order of preference:

1. More partitions, so each task handles less. This is usually the right answer, costs nothing, and scales.

spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB")
spark.conf.set("spark.sql.shuffle.partitions", "2000")

1. Check for skew. One oversized partition spills while its peers do not, and no amount of memory fixes a distribution problem.
2. Fewer cores per executor, which gives each running task a larger slice of the same pool. Going from 5 cores to 3 raises per-task memory by two thirds.
3. More executor memory, last, because it treats the symptom and costs cluster capacity.

Context worth adding: some spill is normal and by design. Spark spills rather than failing, which is a feature. The question is proportion: spill of a few percent of shuffle size is fine, spill several times the shuffle size means you are paying disk I/O for work that should have been in memory.

The job is spending its time on the driver before tasks start. Three usual causes:

1. Too many input files. Planning must list them and compute splits, which on object storage is many API calls. Tens of thousands of small files can take minutes before a single task runs.

print(spark.read.parquet(path).inputFiles().__len__())

2. A deep plan. Hundreds of chained withColumn calls, or a loop that rebuilds a DataFrame, produce a very deep tree that Catalyst walks repeatedly. Collapse them into one select, or checkpoint to truncate.

3. Very wide schemas. Thousands of columns, or deeply nested structs, make every analysis and optimisation pass expensive.

How to tell it apart from execution slowness: in the Jobs tab, look at the gap between the job being submitted and its first stage starting. In the SQL tab, a long total duration with short stage durations points the same way.

The extreme version is a StackOverflowError from the recursive tree walk, which is the same problem at a larger scale.

The related structural fix for the file-count case is a lakehouse format, because Hudi, Iceberg and Delta keep file metadata in their own manifests rather than requiring a directory listing, which is one of the main reasons they exist.

Too many partitions, so per-task overhead dominates the actual work.

Each task carries real fixed costs: serialization of the task description, scheduling on the driver, JVM thread setup, reading a partition’s metadata, and reporting results. At a few milliseconds of work per task, those costs are the job.

It also loads the driver, which must track every task’s state and metrics. A job with hundreds of thousands of tasks can exhaust driver memory on bookkeeping alone.

The fixes depend on where the partitions came from:

With AQE on, coalescing should already be merging small shuffle partitions toward spark.sql.adaptive.advisoryPartitionSizeInBytes, default 64MB. So if you are seeing thousands of tiny tasks after a shuffle, check that AQE is enabled and that coalescing is not disabled.

The target to state: partitions in the low hundreds of megabytes, and a total that is a small multiple of your core count. Tasks lasting well under a second are a signal you have overshot.

Favour several medium executors over a few enormous ones or many tiny ones.

A common starting point is 4 to 5 cores per executor, and there are two reasons:

1. HDFS client contention. Concurrent tasks in one JVM share the same client, and throughput stops improving past roughly that point.
2. Garbage collection. A very large heap makes full collections slow, and a GC pause stalls every task on that executor.

Then memory follows from partition size:

per-task execution memory ~= (heap - 300MB) x 0.6 / cores

Size the heap so that is comfortably larger than your largest partition’s working set, and add spark.executor.memoryOverhead for off-heap, Python workers and native libraries.

--executor-cores 4 --executor-memory 16g --conf spark.executor.memoryOverhead=3g

The failure modes at each extreme:

And state the caveat: these are starting points, not laws. The right answer depends on your partition sizes and workload, and the honest thing to say is that you would start here, measure spill and GC, and adjust.

Pick one. Setting both is contradictory, and Spark resolves it by treating the static number as merely the initial size.

Use static sizing when the job’s parallelism is roughly constant, you know the shape from experience, and predictable performance matters more than cluster efficiency. A nightly batch job is the archetype.

Use dynamic allocation when parallelism varies a lot between stages, the application is long-lived and often idle (a notebook, a Thrift server), or the cluster is shared and holding idle executors is antisocial.

--conf spark.dynamicAllocation.enabled=true --conf spark.dynamicAllocation.minExecutors=2 --conf spark.dynamicAllocation.maxExecutors=200

The requirement to name: an idle executor may still hold shuffle blocks a later stage needs. So dynamic allocation needs either an external shuffle service, or spark.dynamicAllocation.shuffleTracking.enabled, which defaults to true and keeps executors holding live shuffle data.

The case against dynamic allocation for short jobs: the ramp-up costs more than the elasticity saves.

The cluster manager killed the JVM because the process total exceeded what was requested, which is heap plus overhead. It is not a JVM OutOfMemoryError, and the distinction drives the fix.

What lives in overhead: off-heap buffers, Netty’s direct buffers for shuffle, the JVM’s metaspace and thread stacks, native libraries, and the Python worker processes for PySpark.

That last one is why PySpark jobs hit this far more often than JVM jobs: Python memory is entirely outside the heap, so a job with heavy Python UDFs can have a perfectly healthy heap and still be killed.

The sequence:

1. Confirm which it is from the message. This determines everything.
2. Raise spark.executor.memoryOverhead, generously for PySpark.
3. Check for skew, because one oversized partition is a common root cause that presents as a memory shortage.
4. Raise partition count so each task’s footprint shrinks.
5. Raise executor memory, last.

The point to make: raising memory first often just moves the failure to a larger dataset next month, whereas fixing the partition size or the skew fixes it structurally.

The default timeout for network interactions between Spark components, 120s. It acts as the default for several more specific timeouts, including RPC lookups and shuffle fetch waits.

spark.conf.get("spark.network.timeout")   # 120s

When raising it is legitimate: genuinely long operations on healthy infrastructure. Very large shuffle fetches, a heavily contended cluster, or cross-availability-zone traffic.

When raising it is the wrong answer, which is most of the time: the timeout is reporting that something is dead or overwhelmed, not causing the problem. An executor stuck in a long garbage collection cannot answer heartbeats; an executor that was killed cannot serve shuffle blocks. Raising the timeout makes the job take longer to notice, not longer to succeed.

So the diagnosis order:

1. Look at the executors tab. Did an executor die, and why?
2. Check GC time. An executor spending 40% of its time in GC will miss heartbeats, and the fix is memory or partition size, not the timeout.
3. Check for container kills, which point at overhead.
4. Only then consider the timeout, and pair it with spark.executor.heartbeatInterval, which must stay well below it.

A related note: spark.executor.heartbeatInterval defaults to 10s, and it must be considerably smaller than the network timeout or executors get declared lost spuriously.

Find out which JAR each class is actually loaded from, rather than reasoning about what should be on the classpath.

--conf "spark.driver.extraJavaOptions=-verbose:class" --conf "spark.executor.extraJavaOptions=-verbose:class"

That prints a line per class load with its source JAR, turning a guess about shading into a direct answer:

[Loaded com.fasterxml.jackson.databind.ObjectMapper from file:/opt/spark/jars/jackson-databind-2.15.2.jar]

The usual causes:

1. Two versions on the classpath, one from Spark’s jars/ and one from your assembly. Spark’s usually wins, which is why your newer version appears absent.
2. A provided dependency not actually provided at runtime.
3. A transitive dependency you did not know you had.

The fixes, in order:

• Shade the conflicting dependency in your assembly, which is the robust answer: relocate the package so there is no collision.
• Align versions with what Spark ships, which is the simplest when possible.
• spark.driver.userClassPathFirst and spark.executor.userClassPathFirst, which put your jars ahead of Spark’s. Effective but risky, because it can break Spark’s own dependencies.
• --jars or --packages for genuinely missing classes.

The habit to convey: -verbose:class converts an argument into a fact. It is noisy, so filter it, but it answers the question definitively.

A recursive walk over a structure deeper than the thread’s stack. In Spark that structure is almost always the query plan or the RDD lineage, not your code’s own recursion.

The plan gets deep from:

• Hundreds of chained withColumn or filter calls.
• A loop that rebuilds a DataFrame each iteration.
• Deeply nested structs or very wide schemas.
• A long iterative lineage with no checkpoint.

The stack trace is the tell: if the frames are Catalyst classes, an Avro or Parquet schema walker, or TreeNode methods, it is the plan.

Two fixes, and the right one depends on which side threw:

1. Raise the stack on the side that threw, as a direct remedy:

--conf "spark.driver.extraJavaOptions=-Xss64m"
--conf "spark.executor.extraJavaOptions=-Xss64m"