Apache Spark Overview — Unified Analytics Engine for Big Data

What Spark actually is

Spark is one execution engine wearing several front doors. SQL, the DataFrame API, Structured Streaming, MLlib and GraphX are not four systems sharing a brand; they all compile down to the same thing - a graph of stages, each stage a set of identical tasks, each task chewing through one partition, with a shuffle wherever rows have to be redistributed across the cluster. Learn that core and the rest of the surface area stops being a list of features and becomes a set of front ends over one machine.

Two properties fall out of the design and account for most of what practitioners find surprising. First, you describe a result rather than a procedure, so what actually runs bears only a loose resemblance to the code you wrote - a gift when the optimizer is smarter than you are, a trap when your mental model of the plan is wrong. Second, work is only ever divided by partition, which means parallelism, memory pressure, skew and shuffle volume are the same question asked four different ways: how is the data split, and how evenly.

This article is the map. Each section sits at one altitude - enough mechanism to reason with, then a pointer to the article that goes deep on it. The last two sections are not pointers: how to read the Spark UI well enough to locate your actual bottleneck, and an honest account of the jobs where Spark is the wrong answer.

Driver, executors, and the cluster manager

Three roles, and they are not symmetrical. The driver is a JVM running your program: it owns the SparkSession, compiles queries, cuts the resulting plan into stages, hands tasks out, and receives results. Executors are JVMs on worker nodes that do nothing but run tasks, hold cached blocks, and serve shuffle files to whoever asks. The cluster manager - YARN, Kubernetes, or Spark's own standalone master - participates in none of the computation; it grants containers with a fixed core and memory allocation and starts executors inside them.

The split is worth memorising because each role fails with its own signature. Driver failures are usually memory pressure on a single process: a collect() of something larger than you thought, a broadcast whose build side was mis-estimated, or a plan over hundreds of thousands of files whose listing alone will not fit. Executor failures are usually the container limit rather than the heap limit - JVM heap plus off-heap plus overhead exceeded what the cluster manager granted, and the container was killed from outside, which is why the executor log frequently ends mid-sentence with no Java exception to read. Cluster-manager problems look nothing like either: the job simply sits with zero running tasks because the queue has no capacity or the pods are Pending, and nothing in the Spark logs will tell you so.

One derived number matters more than the rest. Executor cores multiplied by executor count is how many tasks can run at any instant. Almost every sizing decision is a comparison between that figure and the partition count of your widest stage. The full component walkthrough is the Spark execution architecture article.

RDDs underneath, DataFrames on top

The RDD is Spark's substrate: an immutable collection sliced into partitions, carrying a recipe for producing each slice and a record of which upstream slices it was derived from. That record is the lineage, and it is why Spark is fault tolerant without replicating anything - a lost partition is not restored from a copy, it is rebuilt by replaying its recipe from durable input. The dependency graph, what recovery costs, and where checkpointing fits are the lineage article.

What you write, though, is almost never an RDD. The DataFrame API - and Dataset, its statically typed sibling on the JVM - is a description of a result over named, typed columns. The distinction is not ergonomic, it is about how much the engine is allowed to know. rdd.map(f) hands the engine an opaque closure: it cannot tell that f touches only two columns, cannot move it past a filter, cannot generate specialised code for it. df.select("a", "b").filter(col("c") > 10) is a data structure the engine can rewrite. Column pruning, predicate pushdown, join strategy selection and code generation are all available only because the DataFrame API refuses to accept arbitrary code.

The rule that follows is to stay declarative as long as you possibly can. A Python UDF is a hole punched through the plan: rows must be serialised out to a Python worker process and read back, and the optimizer treats whatever comes out as unknown. Reach for the built-in functions first, a pandas_udf second - it moves columnar batches over Arrow instead of rows over a pipe - and a plain row-at-a-time UDF only when nothing else expresses the logic.

Nothing runs until an action

Transformations are lazy. filter, select, join, groupBy and withColumn return a new DataFrame and execute nothing; they extend a description. Only an action - count, collect, show, write, take, foreach - forces the engine to compile that description and run it. Laziness is not a performance tweak bolted on afterwards; it is the precondition for optimization. An engine that executed each call as it arrived could never push a filter into a scan, because by the time it saw the filter the file would already have been read.

Three consequences trip people up, and they are really one consequence seen from three angles.

The stack trace points at the action, not at the mistake. The line that throws is df.show(); the bad cast was written forty lines earlier. Debug from the plan, not the line number.

A timer around a transformation measures nothing. Wrapping df2 = df.filter(...) in a stopwatch times the construction of a tree node. Whatever you meant to benchmark happens later, inside the action, mixed in with everything else the action triggers.

Every action re-runs the whole graph. Two count() calls on the same DataFrame read the source twice, shuffle twice and cost twice, unless you explicitly told Spark to keep the intermediate. This is the reason caching exists, and the reason caching is a real decision rather than a free win. To see what an action will actually do before paying for it, read the EXPLAIN plans article.

Partitions, narrow and wide, and the stage boundary

The partition is the only unit Spark parallelises over. One task processes exactly one partition; a stage with 200 partitions submits 200 tasks; a cluster offering 100 task slots runs them in two waves. Every parallelism lever you have is a way of changing how many partitions exist or how evenly rows are spread across them: spark.sql.files.maxPartitionBytes on the read side, spark.sql.shuffle.partitions after a shuffle - it defaults to 200, which is wrong for most clusters in one direction or the other - and repartition or coalesce when you want to say so explicitly.

Dependencies between partitions come in two flavours. A narrow dependency is one where an output partition can be produced from input that is already local to it - map, filter, select - so Spark fuses a whole run of them into a single task that streams one partition through every step without touching the network. A wide dependency is one where producing an output partition requires rows drawn from every input partition, because the data is being regrouped by key: groupBy, join, distinct, repartition. That regrouping is the shuffle, and the shuffle is where a stage ends. The upstream stage must finish writing all of its output before any downstream task may begin reading, which makes it a hard barrier: one straggling task holds the entire cluster.

Which is why a job's cost is mostly the number and size of its shuffles. The write path, spill behaviour and push-based variant are the shuffle article; the two standard ways to avoid one entirely are broadcasting the small side of a join and pruning partitions so the rows never get read.

Why the DataFrame API beats hand-written RDD code

Three separate machines sit between your DataFrame and the tasks that eventually run, and not one of them can do anything for an opaque closure.

Catalyst is the query compiler. It parses the description into a logical plan, resolves names against the catalog, then rewrites the tree with rules: drop columns nobody reads, push predicates down to the scan so a Parquet reader can skip whole row groups, fold constants, collapse adjacent projections, reorder joins using whatever statistics exist. What comes out is a physical plan - concrete operator choices, including which join implementation to use. Details in the Catalyst article.

Tungsten is the representation and execution layer. Rows do not live as JVM objects; they live in a compact binary layout where reading a field is a pointer offset rather than a chase through object headers and boxed types. On top of that, whole-stage code generation collapses an entire operator chain into one generated Java method with a tight loop, eliminating the per-row virtual call that an interpreting operator tree would pay at every step. See Tungsten and whole-stage codegen; the vectorised commercial successor to that path is Photon.

Adaptive Query Execution is the third, and it is an admission that static planning has limits. Once a shuffle completes, Spark holds real statistics - the actual byte size of every output partition - and re-plans with them: coalescing hundreds of tiny partitions into a sane number, splitting a skewed one across several tasks, or demoting a sort-merge join to a broadcast because the build side turned out small after filtering. That is the AQE article.

None of the three reaches inside a UDF or an rdd.map. That is the entire argument for staying on the declarative API, and it is why hand-tuned RDD code that beat DataFrames in 2015 loses to them now.

Memory, and when caching actually pays

An executor's heap is divided into two pools that share one budget. Execution memory holds what operators need while they run: hash tables for aggregation and joins, sort buffers, shuffle assembly buffers. Storage memory holds cached blocks. The boundary between them is not fixed - either side may borrow from the other when it is idle - but the relationship is asymmetric in a way that matters: execution can evict borrowed storage, storage can never evict execution. Cached data is expendable by definition; a half-finished sort is not. The mechanics, including the reserved fraction neither pool may touch, are the memory management article.

When execution memory runs short the operator spills: part of its working set is serialised to local disk and read back later. The job does not fail, it just gets slower - sometimes by an order of magnitude - and the only place that shows up is the spill columns in the stage metrics, which is why plenty of teams pay for it for months without noticing.

Caching is where overviews usually cheerlead, so be blunt about it. Persisting a DataFrame pays only when the same intermediate is consumed by more than one action, and when what you avoid recomputing costs more than what you give up. What you give up is concrete: cached blocks occupy memory that execution wanted, so an enthusiastic cache can push the very stage you were trying to accelerate into spilling. The test is arithmetic - how many times is this reused, what does one recomputation cost, and does the cached footprint fit alongside the working set of the stages that read it. One reuse of a cheap columnar scan is not worth a cache entry; ten iterations over a filtered and joined intermediate clearly is. And MEMORY_AND_DISK is not free insurance: a disk-resident cached block that must be read and deserialised can easily cost more than re-reading the original Parquet with pushdown.

Where the driver runs, and what that decides

Deployment varies along two independent axes, and conflating them causes a lot of confusion.

The first is the resource manager. Standalone is Spark's own: simple, no other tenants, fine for a dedicated cluster. YARN is the Hadoop-era default and still what most on-premise deployments run, with queues and capacity scheduling shared across engines. Kubernetes is where new deployments go - the driver is a pod, executors are pods it requests directly from the API server, and a container image replaces the cluster-wide Spark installation, which makes per-job dependency versions tractable for the first time.

The second axis is where the driver process itself lives, and it is the one that bites. In client mode the driver runs wherever you typed spark-submit - your laptop, a gateway host, a notebook server. That is what you want interactively, because output comes back to your terminal, but it makes your session load-bearing: close the laptop and the application dies, and everything you collect() crosses the network to a machine outside the cluster. In cluster mode the driver is launched inside the cluster as one more container. Submission returns immediately, the job outlives your connection, and driver-to-executor traffic stays on the fast internal network. Production is cluster mode; "the job died when my VPN dropped" is client mode.

Two adjacent capabilities are worth knowing exist. Dynamic allocation lets an application surrender idle executors and request more as the task backlog grows, which only works safely if shuffle files outlive the executors that wrote them - hence the external shuffle service. Spark Connect goes further and splits the client from the driver entirely, putting a gRPC boundary between them so a thin client can drive a long-lived server-side session without embedding Spark at all.

Streaming is the same engine over an unbounded table

Structured Streaming's central claim is that you should not have to learn a second programming model. A stream is a table that keeps growing. A streaming query is the query you would have written against that table. The engine's job is to produce, incrementally and continuously, the answer that the batch query would have given had it run over everything received so far. Same DataFrame API, same Catalyst plan, same stages and tasks.

Execution is micro-batch by default. On each trigger the engine asks the source how far it has advanced - Kafka offsets, a directory listing - plans a batch over just that new range, runs it as an ordinary Spark job, updates state, and commits the new offsets to a checkpoint. That is exactly why the latency floor is roughly one trigger interval plus one job's worth of planning and scheduling: hundreds of milliseconds at absolute best, seconds in ordinary practice. A continuous-processing mode with long-lived tasks exists for lower latency at the cost of a much narrower feature set.

What genuinely differs from batch is that intermediate results must survive between batches. Aggregations, deduplication and stream-stream joins keep a state store - the stock provider holds state on the JVM heap and checkpoints it to the filesystem, with a RocksDB-backed provider available and generally the right choice once state outgrows the heap - and state that is never released grows without bound. That is the whole reason watermarks exist: a watermark declares how late an event may arrive and still be counted, which is what lets the engine decide that a window is final and its state can be dropped. Setting that bound wrong is the most common structured streaming failure, and it is a correctness bug before it is a memory one. See watermarks and late data, state and checkpointing, and the Structured Streaming overview.

Reading the Spark UI to find your bottleneck

Most Spark tuning advice is wasted because it is applied to the wrong stage. The UI will tell you which stage matters and what is wrong with it, but only if you read it in a particular order. What follows is that order.

Start on the Jobs tab, not Stages. Find the one or two stages that own most of the wall clock. Spark jobs are almost always dominated by a small number of stages; an hour spent optimising anything else is an hour spent making a rounding error smaller. The event timeline also shows gaps with no running tasks at all - those are the driver working alone, and they are invisible in every task-level metric.

Then read the summary metrics table for that stage. It gives min, 25th percentile, median, 75th and max for each per-task metric, and the single most informative number in the entire UI is the ratio of max duration to median duration. Near 1 means the stage is uniformly slow: you are genuinely compute- or IO-bound, and more parallelism or less work per row will help. Above roughly 5 means skew: a handful of partitions carry a disproportionate share of the rows, the stage finishes when the largest one does, and adding executors will change nothing at all. Those two diagnoses have completely disjoint fixes, which is why guessing between them is expensive.

In the same table, check spill. Non-zero Spill (Disk) means tasks exceeded the execution memory they were given and paid to serialise their working set out and back. The instinctive fix is a bigger executor; the usually better fix is more partitions, because that shrinks each task's working set without buying hardware. Also compare Shuffle Read Size against input size - a stage reading far more than the job's source data is doing a shuffle you did not intend.

Check task granularity. The task table breaks duration into scheduler delay, deserialization time, executor compute time, and GC time. If scheduler delay and deserialization are a meaningful fraction, your tasks are too small - thousands of tasks each finishing in under 100ms means launch overhead is your bottleneck and coalescing is the answer. If GC time is above roughly a tenth of compute time on an executor, the heap is under pressure and you are probably caching too aggressively.

Use the SQL tab to check what the optimizer actually did. This is the step people skip. Every node in the plan carries actual row and byte counts, so you can compare them against your assumptions. The two classic findings: a filter you were certain was pushed into the scan shows output rows equal to input rows, meaning it ran afterwards and you read the whole table; and a join you assumed was a broadcast appears as a SortMergeJoin because the estimated build side exceeded the threshold. Both are invisible from the Stages tab and obvious here.

Finally, the Executors tab, for cluster-level asymmetry. A stage that is balanced at the partition level can still be lopsided at the executor level through poor locality or an unlucky assignment. Uneven completed-task counts, one executor with far more shuffle read, or a single node with elevated GC usually means the problem is placement rather than data.

When Spark is the wrong tool

Spark's reputation was built on problems that did not fit on one machine. One machine got much bigger since then, and a great deal of Spark in production is now paying distribution costs for data that never needed distributing.

Small data. Every Spark job pays a floor that has nothing to do with data volume: JVM startup, executor acquisition from the cluster manager, plan compilation and code generation, and shuffle output that goes through local disk regardless of how little there is. A job over a few gigabytes routinely loses to a single-process engine reading the same Parquet files with no coordination at all. A useful heuristic: if the working set fits comfortably in one modern machine's RAM - and that is now hundreds of gigabytes - the single-process version will usually win, and it will win on debuggability too. Spark starts earning its overhead when the data does not fit, when a job runs long enough that mid-flight machine failure is likely, or when the source is already spread across a cluster and pulling it to one node is itself the expensive part.

Low-latency serving. Spark's control path is built for throughput and makes no apology for it. Tasks are dispatched on a scale of tens of milliseconds, a query goes through parsing, analysis, optimization and code generation before a single byte moves, and the entire design assumes a job lives for seconds to hours. There is no Spark equivalent of a point lookup that a key-value store answers in a millisecond, and putting Spark behind a user-facing request path means inheriting a latency floor you cannot tune away. Structured Streaming does not change this: micro-batch latency is bounded below by the trigger interval plus batch execution. If you need single-digit-millisecond stream processing, that is Flink's territory; if you need point reads, precompute into a store designed for them.

Iterative work that fits on one machine. Every iteration that ends in a shuffle pays a full synchronisation barrier, and a barrier costs the same whether the payload is a terabyte or a kilobyte. Optimisation loops with small parameter vectors spend more time coordinating than computing, and the collect-update-broadcast pattern funnels each round through the driver, a single process. A tuned single-node numeric library backed by BLAS, or a GPU, will beat a Spark cluster by orders of magnitude on the same arithmetic. Spark's real contribution to machine learning is feature engineering over data too large for one machine - and then handing a small matrix to something else. Where genuinely distributed iterative training is required, Spark supports it through barrier execution mode, which exists precisely because the default task model does not fit that shape.

Two more that show up constantly. Row-at-a-time logic that cannot be expressed declaratively pushes you into UDFs, and enough UDFs turn Spark into a slow distributed for loop with none of the optimizer's advantages. And high-frequency small writes are a poor fit for a system whose output unit is a file per task - the classic small-file explosion, which is much of why table formats like Delta Lake exist.

The honest failure mode is none of these individually. It is that the cluster already exists, so every new problem gets solved on it. Sizing the tool to the problem is worth doing deliberately, once, per workload.