Apache Spark Conceptual & Architecture

The reasoning

The driver hosts the SparkContext, builds the logical and physical plans, splits the job into stages at shuffle boundaries, schedules tasks onto executors, tracks which succeeded, and receives whatever an action returns. It is a single process and it is a single point of failure — if it dies the application dies, and any cached data on the executors goes with it.

Executors are JVM processes that do the work. Each runs tasks in slots — one task per core — holds cached partitions in its storage memory, and writes shuffle files to local disk. They are horizontally scalable and individually replaceable: losing one costs you the recomputation of whatever it held.

The failure modes divide cleanly and naming them is what shows understanding. Driver OOM comes from pulling data to it — collect() on a large DataFrame, toPandas(), or a broadcast whose build side turned out bigger than the estimate. Executor OOM comes from a single partition not fitting in a task's memory, which is almost always skew. If someone reports 'Spark ran out of memory', the first question is which process, because the two have nothing to do with each other.

The reasoning

A job is what an action creates. count(), collect(), write() — each triggers one job, which is why a script with four writes shows four jobs in the UI even though you wrote one pipeline.

A stage is a span of work with no shuffle in it. The scheduler cuts the plan at every Exchange, because a stage's tasks must be able to run independently and a shuffle requires all of the previous stage's output. That is the mechanical link between the plan and the UI: stages are the segments between Exchange nodes, and the whole-stage codegen markers *(1), *(2) in a plan number them.

A task is one partition of one stage — the unit actually scheduled onto a core. So task count comes from partition count, and partition count has different sources at different points: on a scan it is the number of input splits, which is roughly file count for modestly-sized files; after a shuffle it is spark.sql.shuffle.partitions, or whatever AQE coalesced it to. Twelve files and 200 shuffle partitions gives roughly 12 tasks then 200.

The reasoning

The test is dependency, not operation name. A transformation is narrow when each output partition depends on exactly one input partition — filter drops rows within a partition, withColumn computes from the row in front of it, select projects. None of them needs to know about a row on another executor, so all three fuse into a single stage.

You can read that off the plan: there is no Exchange, and the codegen markers show one group, meaning Spark compiled the whole chain into a single generated function operating on a row at a time. That fusion is why chaining twenty narrow transformations costs almost nothing beyond the one you needed.

Wide transformations are the ones where an output partition needs input from many partitions — groupBy, join on a non-broadcast side, distinct, orderBy, window, repartition. Each forces a shuffle, and therefore a stage boundary. The practical form of this knowledge is that you can look at a chain of PySpark and count the stages before running it, which is exactly what a plan-reading question asks you to do.

The reasoning

Read it bottom-up, which is how a physical plan is written. The scan, the first withColumn and the partial aggregate are all below the Exchange — one stage, fused into one generated function. The Exchange is the boundary. The final aggregate and the second withColumn are above it, in the next stage.

That placement has a real consequence: work below the shuffle happens on the pre-aggregation row count, and work above it happens on the aggregated row count. Here that is 2,000 rows versus 3. Expensive per-row work is therefore much cheaper above a shuffle than below it, and moving a computation across that line is a legitimate optimisation.

The partial-then-final aggregation is worth naming while it is on screen. Spark does not shuffle raw rows to group them — it aggregates within each partition first, shuffles the partial results, and combines. That is why a sum over billions of rows shuffles almost nothing, and why an aggregation with high cardinality keys is far more expensive than one with three regions.

The reasoning

Nothing executed while the loop ran. Each withColumn added a node to a logical plan, Catalyst collapsed the chain, and the projection pruning pass removed the expressions whose outputs the select does not keep. The plan shows a single Project with the surviving expressions in it.

That is the practical face of laziness: you can build a plan across many lines and even many functions, and pay only for what the final action needs. In pandas the same loop allocates five new columns of real memory immediately, whether or not you use them.

The trap worth knowing is that laziness has a cost of its own. Long chains build large plan trees, and Catalyst's analysis is not free — thousands of iterative withColumn calls in a loop is a well-known way to spend minutes in *planning* before a single task runs. The fix is a single select with a list of expressions rather than a loop of withColumn, which produces the same plan without building and rewriting it a thousand times.

The reasoning

Catalyst combines adjacent filters and pushes the result as far toward the source as it can. Both predicates end up in the scan's PushedFilters, which means the Parquet reader uses row-group statistics to skip data that cannot match — the rows are never decoded.

So micro-optimising filter order is wasted effort for simple predicates, and that is worth knowing because a lot of hand-tuning goes into it. What is *not* wasted is knowing where the optimizer stops: a Python UDF is opaque and cannot be pushed; a predicate on the null-producing side of an outer join cannot be moved below the join without changing the result; and a filter on a column produced by a window cannot move below the window at all.

The other half of the answer is which predicates Parquet can actually use. Equality and range comparisons on primitive columns work well. A filter needing string manipulation, a cast, or a function over the column may be pushed as a DataFilter but not usable as a PushedFilter — the plan distinguishes them, and reading both lines is how you tell whether pushdown really happened.

The reasoning

An action is anything that needs a result: count returns a number, collect returns every row to the driver, show runs a limited job and prints, and write runs the whole plan and produces files. Each creates a separate job, which is why the Spark UI shows more jobs than you have pipelines.

cache() is the one people get wrong. It is a *marker* — it registers the intent to persist and returns immediately. Nothing is stored until an action runs the plan, and even then only the partitions that action touched. df.cache() followed by nothing has done precisely nothing, and df.cache(); df.count() is the idiom because the count is what actually materialises it.

explain() is the other one: it runs analysis and optimisation and prints the resulting plan, without scheduling a single task. That is what makes it safe to call on anything, and it is the reason plan reading is a cheap skill to practise — you can inspect what a job would do without paying for it. The distinction to state clearly is between planning, which is driver-side and free, and execution, which is the cluster.

The reasoning

Cores are task slots: 5 cores means 5 tasks in flight on that executor. Memory is not divided per task — it is a pool. Of the 20 GB a slice is reserved, and spark.memory.fraction (0.6 by default) of the rest is a unified region shared between execution and storage, borrowed dynamically by whichever tasks need it.

That sharing is what turns skew into an OOM rather than a slowdown. Four tasks holding small partitions and one holding a huge one is not a balanced split of the pool: the large task takes what it can, spills the rest to disk, and if a single group cannot fit even alone it fails outright. Spill is the graceful version; OOM is what happens when spilling cannot save you.

The sizing trade is between few fat executors and many thin ones. Fat executors share one broadcast copy and give large tasks room, and they suffer longer garbage collection pauses and lose more computed work when one dies. Thin executors recover faster and each holds its own copy of every broadcast. Around 4 to 5 cores per executor is the usual recommendation because per-process I/O throughput stops improving much past that — and the answer worth giving is the reasoning, not the number.

The reasoning

The build side is collected to the driver first. That is the step people forget, and it is why an oversized broadcast produces a driver OOM rather than an executor one — the driver holds the whole thing before it can distribute anything. Then it is sent to each executor, which builds a hash table and keeps it for the job's duration.

So the total is one copy on the driver plus one per executor, and the deserialised hash table is typically larger than the compressed Parquet the size estimate came from. A dimension that looks like 200 MB on disk can be several times that as a hash map of objects, which is how a broadcast that passed the threshold still exhausts an executor.

The symptoms are distinctive enough to name. spark.sql.broadcastTimeout exceeded means the collect-and-ship did not finish in time, usually because the side is far bigger than assumed. A driver OOM in a job with a broadcast hint is the same cause one step earlier. And if neither fires but every executor is under memory pressure, the broadcast worked and is simply too expensive to hold fifty times.

The reasoning

The routine rewrites: predicate pushdown so filters reach the scan, column pruning so only referenced columns are read, constant folding so WHERE 1=1 AND x>5 becomes x>5, null propagation including the isnotnull filters inserted before join keys, and join reordering when statistics exist to reorder on.

The boundaries are where your work is. A Python UDF is a black box: it cannot be pushed, reordered or reasoned about, and its presence disables optimisation around it. An aggregation is not moved through a join in the general case, because that is only valid under conditions the optimizer usually cannot prove — so aggregate-before-join is a rewrite you write yourself. And Catalyst optimises one query at a time: it will not notice two branches reading the same expensive intermediate, which is why cache() exists.

The other limit is statistics. Cost-based decisions — join order, join strategy — are only as good as the estimates, and without ANALYZE TABLE those come from file sizes. Compressed Parquet under-reports in-memory size, so the estimate is systematically low. That is the strongest argument for AQE: it replaces guesses with measurements at each stage boundary.

The reasoning

The parsed logical plan is pure syntax: it knows you wrote a select and a filter and has no idea whether custmer_id exists. Analysis resolves every attribute and function against the catalogue, which is where AnalysisException comes from — a typo, a column made ambiguous by a join, a function that does not exist. It is driver-side and instant.

Optimisation applies the rule-based rewrites to produce the optimised logical plan. Physical planning then turns logical operators into strategies: a logical Join becomes BroadcastHashJoin or SortMergeJoin, a logical Aggregate becomes HashAggregate with an Exchange. That last step is where statistics and configuration enter.

Practically this is why explain() is a fast feedback loop. It runs the whole front half of the compiler without scheduling work, so it catches name errors, shows the join strategy, and tells you whether the filter reached the scan — for the cost of a driver-side call. Printing a plan before running an expensive job is one of the cheapest habits available.

The reasoning

The write side: every task in the upstream stage computes a partition id per row, buffers rows by destination, spills to local disk when the buffer fills, and writes one shuffle data file plus an index. Those files are on the executor's local disk, not in the distributed store — which is the fact that explains the most confusing shuffle failures.

The read side: each task in the downstream stage fetches its partition's bytes from every upstream task. That is an M x N fetch pattern, so connection count grows with the product of the two stage widths rather than their sum. It is why very high shuffle partition counts hurt even when each partition is tiny.

The failure modes follow directly. FetchFailedException means the shuffle files were unreachable, usually because the executor holding them died or was reclaimed — and since the files were on its local disk they went with it, forcing the whole upstream stage to be recomputed. Disk pressure on executors is shuffle spill. And an executor lost during a long shuffle can cascade, because the recomputation shuffles again. The external shuffle service exists precisely so those files outlive the executor that wrote them.

The reasoning

repartition performs a full shuffle, so it can raise or lower the count and produces even partitions — including a hash-partitioned form when given columns. coalesce avoids the shuffle by merging existing partitions into fewer, which is much cheaper and can leave the result uneven, because it combines whatever was there.

The second difference is the dangerous one. Because coalesce introduces no stage boundary, the reduced parallelism applies to the whole stage it sits in — it propagates upstream. df.map(expensive).coalesce(1).write() does not run the expensive work in parallel and then merge; it runs the entire stage with one task. If the goal is one output file after expensive work, repartition(1) is correct despite the shuffle, because the shuffle creates the boundary that keeps the upstream parallel.

So: coalesce to reduce output files after cheap work; repartition when you need evenness, an increase, a specific distribution, or a stage boundary protecting upstream parallelism. On modern Spark, AQE's partition coalescing handles the common post-shuffle case automatically, which removes most of the reasons people reached for either.

The reasoning

The definition is about dependency, not about the operation. A join is wide when matching keys have to be brought together, and it is not wide when they already are — or when one side is available everywhere.

Broadcast is the everywhere case. The build side is collected and shipped to every executor, so each partition of the large side joins without moving. The plan shows BroadcastHashJoin with no Exchange, which is why a six-dimension star-schema query can still have exactly one shuffle.

Bucketing is the already-together case, and it is the more interesting answer. If both tables are written bucketed by the join key into the same number of buckets, matching keys are already in corresponding files and Spark can sort-merge join with no Exchange on either side. That is the durable version of the repartition trick: you pay the shuffle once at write time and every later join on that key is free. The cost is that bucketing constrains the write path and fixes the bucket count, which is why it is worth it for a key joined the same way many times a day and not otherwise.

The reasoning

Catalyst is the reason people give. A DataFrame expression is declarative, so the optimizer can push filters, prune columns, reorder joins and choose strategies. An RDD map is an opaque function: Spark runs it and can reason about nothing.

Tungsten is the second, and it is about representation. DataFrame rows live in a compact binary format Spark manages directly, so there is no per-object JVM overhead, garbage collection pressure is far lower, and whole-stage code generation can compile a chain of operators into one tight loop over that memory. RDDs of Java or Python objects get none of it.

The third is specific to PySpark and usually the largest. DataFrame operations are descriptions executed entirely in the JVM — your Python process builds a plan and waits. An RDD lambda has to run in Python, so every row is serialised out of the JVM, processed, and serialised back. That is why the gap between the APIs is far wider in PySpark than in Scala, and why 'drop to an RDD for control' is much more expensive advice in Python.

The reasoning

A Dataset is a typed collection: Dataset[Order] knows its element type at compile time, so the compiler rejects a reference to a field the case class lacks. That needs static types and a generated encoder, neither of which Python has. In Scala, DataFrame is literally a type alias for Dataset[Row] — the untyped end of one API.

So in PySpark, errors Scala catches at compile time surface at analysis time instead: an AnalysisException when the plan is resolved, before any task runs. That is a real difference in feedback speed and a small one in practice, because analysis is instant and happens on the driver.

The thing worth knowing is that typed Datasets are not automatically faster in Scala either. A typed lambda in ds.filter(o => o.amount > 100) is a closure Catalyst cannot see into, so it loses pushdown exactly as a UDF does — the expression form optimises better. Type safety and optimisability pull in opposite directions, which is not what most people expect.

The reasoning

The clearest case is a custom partitioner. The DataFrame API offers hash and range partitioning; if you need rows grouped by something neither expresses — a locality rule, a hash chosen to co-locate with an external system — rdd.partitionBy is the only route. Even then the usual answer is to compute a partition-key column and hash-partition on that instead.

The second is genuinely unstructured input: a log format needing stateful parsing across lines, or a binary format with no reader. mapPartitions gives you a whole partition at a time to parse and then convert to a DataFrame — and note the shape of that advice, which is to drop down for the parse and come straight back up.

The cost is everything the structured API buys: no pushdown, no pruning, no whole-stage codegen, objects instead of Tungsten rows, and from PySpark a serialisation round trip per row. mapPartitions at least amortises Python startup across a partition rather than a row, which is why it is preferred over map whenever you are down there. If someone reaches for df.rdd.map to apply row-level logic, the answer is nearly always a native expression or a pandas UDF.

The reasoning

A native expression is compiled into the generated JVM code, can be pushed into the scan, and never leaves the JVM. If the logic is expressible with column functions — and when/otherwise, regexp_extract, transform and the higher-order functions cover far more than people try — this is the answer.

A scalar Python UDF is the opposite end: rows are serialised out of the JVM, deserialised into Python objects, the function is called per row, and results go back. On top of that cost the plan shows BatchEvalPython, and the scan can no longer push your predicate — so you also read more data than you needed to.

A pandas UDF sits between them, and the improvement is specifically about transport. Data moves as Arrow batches, the function receives a pandas Series and returns one, so per-row call overhead and most of the serialisation cost disappear — often several times faster. What it does not fix is opacity: Catalyst still cannot see the logic, so pushdown and reordering are still lost. The ordering is native, then pandas UDF, then scalar UDF, and the jump from the last to the middle is usually larger than people expect.

The reasoning

Spark tracks the lineage of every partition — the chain of transformations that produces it — so a lost partition is recomputed rather than replicated. The tasks that were running on the dead executor are rescheduled, and the driver marks its results as gone.

The cost depends entirely on what was lost. Tasks merely in flight are cheap: rerun them. Shuffle files the executor had already written are expensive, because a downstream stage needs them and they lived on that executor's local disk — so Spark reports FetchFailedException and re-runs the map tasks that produced them, which can cascade back several stages. Cached partitions are somewhere between: recomputed on next use, and cheap only if the lineage behind them is.

Two things change that arithmetic and are worth naming. An external shuffle service holds shuffle files outside the executor process, so losing an executor no longer loses its shuffle output — which is why it is close to mandatory on spot instances or with dynamic allocation. And checkpointing truncates lineage by writing to reliable storage, which is how you stop a very long chain from being recomputed from the source.

The reasoning

CSV has no types. To produce a DataFrame with a schema, Spark has to look at the data — with inferSchema it samples or scans the file, decides each column's type, and only then plans your query. On a large file that is a full extra pass, and it happens on every run.

It is also a correctness risk, which is the part worth raising. Inference is a guess from what it saw: a column of digits that is really an identifier becomes an integer and loses leading zeros; a column that is integral in the sampled rows and decimal later fails or coerces. Supplying an explicit StructType removes both the pass and the guess, and it makes a schema change from upstream fail loudly instead of silently retyping.

Parquet has none of this because the schema is in the file footer. Reading it is a metadata operation, the types are exact, and the same footer carries the column statistics that make predicate pushdown work. That is the practical argument for converting text sources to Parquet at landing and never reading the raw CSV twice.

The reasoning

The driver hosts the SparkSession, builds the logical and physical plan, cuts the plan into stages at shuffle boundaries, schedules tasks onto executors, tracks which succeeded, and receives whatever an action returns. It is one process, so it is one point of failure.

The cluster manager — YARN, Kubernetes or standalone — allocates the containers the executors run in. It is the component candidates drop, and dropping it is where the common confusion starts: the cluster manager decides how much hardware the application gets, and it knows nothing at all about stages, tasks or shuffles. It cannot schedule your work and it cannot tell you why a stage is slow.

Executors are the JVM processes that do the work. Each runs tasks in slots — one task per core — holds cached partitions in storage memory, and writes shuffle files to local disk. They are replaceable: losing one costs the recomputation of whatever it held, not the job.

The reason this ordering matters is that it tells you where to look when something breaks. A resource request that never gets granted is a cluster-manager problem, a plan that schedules badly is a driver problem, and a task that dies is an executor problem — and the logs live in three different places accordingly.

The reasoning

SparkContext was the original entry point. It creates RDDs, accumulators and broadcast variables, and it is the object that actually holds the connection to the cluster. Every Spark 1.x example starts by constructing one, and older code often also carries a SQLContext or HiveContext alongside it.

Spark 2.0 collapsed those into SparkSession, which is what you build today: DataFrames, Spark SQL, Structured Streaming and the catalog all hang off it. It does not replace SparkContext so much as contain one — `spark.sparkContext` returns it, and that is not a legacy escape hatch, it is the supported way to reach the APIs that never moved.

So the useful version of the answer is: build a SparkSession, and drop to `spark.sparkContext` for accumulators, broadcast variables, `parallelize`, and setting a job group or job description. Saying "SparkContext is deprecated" is wrong — it is not deprecated, it is wrapped.

The reasoning

Every transformation you call adds a node describing an operation and its inputs. Nothing executes: the driver is building a graph. It is directed because data flows one way and acyclic because an operation never feeds back into its own input, which is what lets the scheduler order it at all.

An action forces evaluation. At that moment the driver optimises the graph, cuts it into stages at every shuffle boundary, and turns each stage into one task per partition. So the unit an action creates is a job, and a job has its own DAG.

Three actions therefore give three jobs and three DAGs — and this is the part that surprises people, because each one starts again from the source. A script that calls `count()`, then `show()`, then `write()` on the same DataFrame reads the input three times and reruns every transformation three times. Caching between them is what makes the second and third jobs cheap, and it is the reason the Spark UI showing four jobs when you wrote "one pipeline" is not a bug.

The reasoning

Every DataFrame or RDD knows the operations that produced it. When an executor dies, the driver does not need a replica — it reschedules the lost partitions and replays the transformations that built them. This is why Spark can run on cheap, preemptible hardware where losing a node is routine.

The cost is that recomputation walks backwards through the whole chain. In an iterative job — a loop that unions or joins onto the same DataFrame each pass — the lineage grows every iteration, so a failure late in the loop replays everything from the start, and the plan itself gets slow to build and can eventually blow the driver's stack.

Checkpointing is the cut. It writes the data to reliable storage and **truncates the lineage**: the checkpointed DataFrame's parents are forgotten, so recovery restarts from the checkpoint instead of the source. That is a different purpose from caching, and it is where the two get confused. Cache is a performance hint — it can be evicted, and if it is, lineage silently rebuilds it. A checkpoint is durability: it costs a full write and a second pass, and in exchange the history is genuinely gone.

The practical rule: cache what you reuse, checkpoint what you cannot afford to recompute or whose lineage is growing without bound.

The reasoning

The four phases. **Unresolved logical plan**: the shape of your query, with column and table names not yet checked. **Analyzed**: names resolved against the catalog, types attached — this is where a misspelled column becomes an AnalysisException. **Optimized logical plan**: rule-based rewrites applied. **Physical plan**: one or more executable strategies generated, costed, and one chosen — this is where broadcast versus sort-merge join is decided.

The rules worth being able to name: **predicate pushdown** moves filters as close to the source as possible, and for Parquet or ORC that means the scan itself skips row groups. **Column pruning** reads only the columns the query references. **Constant folding** evaluates constant expressions once — `price * (10/2)` becomes `price * 5`. **Join reordering** and **boolean simplification** rearrange the query shape. **Limit pushdown** stops a scan early.

The distinction to hold onto is that the first three phases are rule-based and the last one is where cost estimates enter. That is why a plan can change without your code changing: statistics changed, so the physical strategy changed. It is also why the optimizer can be wrong — the rules are always safe, the cost estimates are guesses, and a stale statistic that says a table is 8 MB will get it broadcast when it is 8 GB.

The reasoning

Without codegen, each operator is an iterator that pulls a row from the one below through a virtual call, boxing and unboxing values as it goes. That is a lot of overhead per row for work that is often a single comparison.

Whole-stage codegen collapses the operators in a stage into one generated function — a tight loop over raw memory, with no virtual dispatch and far less object churn, which also keeps the CPU cache useful. You can see it in a physical plan: operators inside a codegen stage are marked with `*` and a stage number, like `*(2) Filter`.

It does not always apply, and the absence of those markers is a real diagnostic signal. Spark falls back when an operator does not implement codegen support, and when the generated code would exceed its size limits — controlled by `spark.sql.codegen.maxFields` (default 100) for very wide schemas, and by the JVM's method size limit, which Spark guards with `spark.sql.codegen.hugeMethodLimit`. A query over a table with hundreds of columns, or one with an enormous chain of expressions, can quietly lose codegen and get slower for no visible reason.

The reasoning

MapReduce runs one fixed shape: map, shuffle, reduce — and it materialises the output of every phase to HDFS. A five-step pipeline is therefore five separate jobs with four full round trips to disk, and an iterative algorithm re-reads its input on every pass.

Spark plans an arbitrary DAG. Intermediate results stay in executor memory unless they spill, stages are cut only where a shuffle genuinely requires it, and consecutive narrow operations fuse into a single pass over the data. For an iterative workload — the case Spark was originally built for — that is the difference between re-reading the input each iteration and keeping it resident.

The second half is scope. MapReduce is a batch execution engine; anything else was a separate system on top (Hive for SQL, Storm for streaming, Mahout for ML). Spark ships one engine with SQL, structured streaming, and ML on the same runtime and the same optimizer, which matters more in practice than the raw speed number.

The claim to be careful with is "100× faster". That figure comes from a specific in-memory iterative benchmark. On a single-pass job that reads a terabyte, filters it and writes it back, both engines are bounded by I/O and the gap is far smaller. Say where the advantage comes from and the number takes care of itself.

The reasoning

**Tumbling**: fixed length, no overlap. Five-minute tumbling gives 10:00–10:05, 10:05–10:10, and every event lands in exactly one. **Sliding**: fixed length, emitted more often than its length — a five-minute window every minute, so an event belongs to five windows at once. More granular and roughly five times the state and compute. **Session**: no fixed length at all; a window closes after a gap of inactivity, which is what you want for user sessions where the natural boundary is behavioural rather than clock-based.

The watermark answers a different question. Windows group by event time, but events arrive late, so at any moment a window might still receive more data. The watermark is a declared bound — "I will not wait for anything more than ten minutes late" — expressed as the maximum event time seen so far minus that threshold. When the watermark passes a window's end, the window is closed, its result emitted, and its state dropped.

This is why the two are inseparable in practice. **Without a watermark, a windowed aggregation over an unbounded stream has no reason to ever finalise a window, so state grows without limit** — the engine cannot know that no more 10:00 events are coming. And the threshold is a real trade: a long watermark waits for stragglers and delays every result; a short one emits promptly and drops the late arrivals, which then have to be handled somewhere else, as a side output or a correction job.