Apache Spark

Suppose a query joins a 100 GB orders table with a much smaller customers table and aggregates revenue by country. On a laptop this takes a long time; on a running Spark cluster it finishes in a minute or two. Yet a job that reads a table with only twelve new rows can take five minutes when its cluster has to start first, because waiting for machines, starting Spark's processes on them, planning the query and publishing the result cost the same whether a job moves 100 GB or a dozen rows. Both behaviors follow from the same design: Spark splits the data into pieces and processes them in parallel on many machines, moves data between machines mainly at a few well-defined points, and pays fixed costs that don't shrink with the data: start-up for every application, and planning, scheduling and publishing the result for every query. Once you see where those costs and those points are, most of Spark's performance and failure behavior becomes predictable.

This post is the entry point to the Spark series. We start with why Spark spreads the work over many machines and the mental model that follows from it, then follow the orders query from DataFrame code down to tasks running on executors and files in a table. Then we look at what surrounds that path: memory and caching, Python, troubleshooting, streaming, deployment and Spark Connect, each a short tour of a later post. After that comes a map of every Spark post in this series and guidance on when Spark is the right engine and when it is not. Everything here describes Spark 4.2.0, the newest feature release when this post was written.

From One Machine to Many#

Before Spark's own vocabulary, consider what limits a query on one machine. Every order the query needs has to be read from storage, decompressed, decoded, matched with its customer and added to its country's sum, and a machine does that work with a fixed number of cores, a fixed amount of memory and one network link. Cut the table into 800 pieces of about 128 MB. A machine with 4 cores processes 4 pieces at a time, so it needs 200 waves to get through them. Ten such machines process 40 pieces at a time and need only 20 waves. Between them, they also have ten network links to read through and ten times the memory.

The query doesn't finish ten times sooner, though. The machines' processes have to start, and the query has to be planned and its 800 pieces handed out; none of this gets shorter as machines are added. Some rows also have to travel between machines, a cost that one machine never pays. For our 100 GB, these costs are a small part of the run; for a few rows, they can be almost all of it, as in the hypothetical twelve-row job from the opening.

A larger machine is the other way to go faster, and up to a point the better one. Cloud providers rent single machines with hundreds of cores and terabytes of memory, and on one of them, an engine built for one machine, such as DuckDB or Polars, often beats a cluster; the section on when to use Spark, near the end of this post, says up to what size. A cluster offers what one machine can't:

• No ceiling. A machine has a largest size; a cluster grows by adding machines, and its cores, memory and network links grow with it.
• Elastic scaling, and costs that follow it. A cluster can be set to grow while the work needs more machines and to shrink when it needs fewer, even within one job: reading our 100 GB keeps every core busy, but a later step that works on much less data needs far fewer. In the cloud, where machines are billed for the time they run, the bill follows the work too. One machine can't resize while a job runs, so it is sized for the heaviest step and paid for at that size throughout.
• Sharing. Many jobs, from many teams, can run on one pool of machines, each taking the share it needs at the time.

These gains come with three problems that one machine doesn't have:

• The work needs a coordinator. The work has to be cut into pieces, each piece sent to a machine with a free core, and the results collected.
• Rows end up on the wrong machine. On one machine, all cores share one memory: whatever one core has loaded or computed, every other core can read. Across machines, no memory is shared: when the work on one machine needs rows that another machine holds, they have to travel over the network, which is far slower than reading memory.
• Machines fail. The more machines a job runs on, the more likely it is that one of them is lost before the job ends, so the job must be able to lose a machine without starting over. Handled well, this becomes a strength: a lost machine usually costs only its share of the work, while a single machine that fails takes the whole job with it.

The mental model in the next section shows Spark's answers: one process coordinates the others, the data is cut into pieces that are processed independently, rows cross machines mainly at a few points, and Spark records how each piece was computed, so that it can compute it again.

The Mental Model#

Four ideas explain most of what Spark does, and the APIs and the relational engine built on top of them explain most of the rest. We introduce each briefly here; the following posts take them apart.

A driver plans, executors execute#

Your program runs in the driver: a JVM process or, for PySpark, your Python process together with a JVM that it controls. The driver builds a description of the computation and, when you ask for a result, turns that description into tasks. Executors are separate JVM processes on worker machines, started for one application (one run of your program, with its driver) and gone when it ends. Each executor has a fixed number of cores (each core is a task slot, a thread that runs one task at a time), a heap (the memory where the JVM keeps its objects), and local disk for the files that shuffles write (described below) and for spill, the data a task writes to disk when it runs short of memory. A third party, the cluster manager (YARN, Kubernetes or Spark's standalone master), starts executors when the driver asks for them; when the driver itself runs on the cluster (cluster mode, described later), the cluster manager starts it too. The cluster manager never decides which task runs where.

Executors never call back into your program on the driver. They receive serialized tasks from the driver, run them, and report results and metrics back. The driver is also the only process that knows where every piece of intermediate data lives (which executor wrote which file of intermediate results, which executor keeps which piece of data in memory for reuse). The consequence for operations is asymmetric: losing an executor usually costs some recomputation, while losing the driver ends the application; at best, the cluster manager starts it again from the beginning.

Data is split into partitions, and a task processes one partition#

A partition is Spark's name for the pieces above (Spark numbers them from 0): a slice of a dataset that one task processes from start to finish. Our 100 GB orders table, read as Parquet with the default 128 MB split size, becomes roughly 800 input partitions, and Spark runs one task per partition, as many at a time as there are free task slots. Our cluster has 10 executors of 4 cores, each with an 8 GB heap: that is 40 tasks at a time, so the scan runs in about 20 waves.

Parallelism, memory per task, the number of files a write produces, and skew are all partition questions. When a Spark job is "slow", the first question is usually whether the partitions are too big (every task spills or runs out of memory), too small (scheduling overhead dominates), or skewed (a few partitions hold far more data than the rest, so a handful of tasks run long after the others have finished).

Shuffles cut the job into stages#

Some operations need all rows with the same key in one place: a join, a GROUP BY, a window function, a DISTINCT. Our query needs every order and the matching customer row on the same executor before it can join them, but the 800 orders partitions were cut by file position, not by customer_id.

Spark solves this with a shuffle. Every upstream task splits its output into buckets, one per downstream partition, by hashing the key, and writes the buckets to local disk. Every downstream task then fetches its bucket from every upstream task, so a shuffle is an all-to-all exchange. The tasks that write a shuffle are its map side, named after the map phase of MapReduce, Hadoop's batch engine that came before Spark, and the tasks that read it are its reduce side. Rows for any key may still be in the last upstream task to finish, so no downstream task can produce a complete result before the whole upstream side is done, and Spark starts no downstream task until the driver has registered the output of every upstream task. Spark therefore cuts the computation into stages at every shuffle: inside a stage, one task runs all of the stage's operations on its partition, passing rows from one to the next; between stages, data is written to local disk and fetched over the network.

Operations that work on each partition alone, such as filter and select, are called narrow. Those that need rows from other partitions, such as join and groupBy, are wide, and they cause a shuffle unless the rows are already split by the key, or one side of a join is copied to every executor, the broadcast join described below. The shuffle is usually the most expensive thing a Spark job does, and it gets a post of its own.

A common misconception follows from Spark's early marketing as an "in-memory" engine. Spark keeps working data in memory where it can, but every shuffle writes its map output to local disk, sorts and aggregations spill to disk when memory runs short, and cached data may live on disk. What Spark avoids is the step between MapReduce jobs. A MapReduce job has one map phase and at most one reduce phase: within it, a filter and a map run together in the mapper, and the map output goes to local disk, much like a Spark stage. But a computation with several shuffles (a join, then an aggregation, then a sort) needs a chain of MapReduce jobs, and each one writes its result to HDFS, Hadoop's distributed file system, which keeps three copies of it by default, for the next job to read back. Spark runs all of those shuffles within one query: the reduce side of one shuffle feeds the map side of the next inside the same task, and only shuffle files on local disk sit between them.

Evaluation is lazy, and lineage provides fault tolerance#

Transformations (filter, join, select) only build a plan; actions (count, collect, saveAsTable) trigger execution. Laziness is what lets Spark see the whole computation before running any of it, and it has a second benefit: Spark records how every partition was derived from its inputs, its lineage, so it can recompute a lost partition instead of replicating data as it goes.

This is the central idea of the 2012 paper on resilient distributed datasets (RDDs), the partitioned, immutable collections that were Spark's first API and still run beneath DataFrames, and it still holds in Spark 4. A failed task is re-run on the same input; a lost shuffle file triggers re-execution of only the map tasks whose output was lost with it (since 4.2, if a re-run map task of a DataFrame or SQL query produces different data, Spark also re-runs the stages that already read the old data); and only the driver is a single point of failure. The cost of this choice is that recovery time depends on how long the lineage is, which is why iterative jobs sometimes need checkpoints, which save a dataset to reliable storage and cut the lineage there.

RDDs, DataFrames and Datasets#

Spark has three APIs for a partitioned dataset: RDDs, DataFrames and Datasets. The RDD came first, and DataFrames and Datasets were built on top of it.

An RDD, the resilient distributed dataset of the paper above, is an immutable collection, split into partitions, that knows how to compute itself. Spark describes every RDD by five properties, which its source code lists in this order:

1. a list of partitions, the slices that tasks process;
2. a compute function, which produces the rows of one partition;
3. a list of dependencies on parent RDDs, its lineage: narrow when each partition needs only a few parent partitions, as after a filter, which reads partition 7 of its parent to produce its own partition 7, and wide when it needs rows from every parent partition, which takes a shuffle;
4. optionally, a partitioner, which says, for an RDD of key-value pairs, which partition each key belongs to, such as the hash partitioner of an RDD that comes out of a shuffle by key;
5. optionally, the preferred locations of each partition, such as the machines that store a file's blocks on HDFS, so that the scheduler can run its task there.

Transformations never change an RDD: filter and map each return a new RDD whose dependency points to its parent. An action hands the last RDD to the scheduler, which cuts the graph into stages at its wide dependencies. A task computes one partition by calling the compute function of its stage's last RDD, which asks its parent for the partition it is computed from (after a filter or a map, the one with the same number), and so on back to the input, so each row passes through all the narrow steps in one go. Once our query appears as DataFrame code, we write it with RDDs too and compare.

A DataFrame adds what the RDD lacks: a schema, with named and typed columns that Spark understands. Its operations don't build RDDs; they build a plan of relational operators, which Spark optimizes, as the next subsection describes. When an action runs, Spark compiles the optimized plan into RDDs, whose partitions hold rows in a compact binary format, also described next. A DataFrame query therefore still runs as an RDD graph, but one that Spark generated, not one you wrote.

A Dataset is the typed form of the same API, in Scala and Java: a Dataset[Order] holds objects of a class Order, and an encoder converts between those objects and Spark's binary rows. Operations written as column expressions, such as ds.filter($"amount" > 100), are optimized like a DataFrame's. Operations written as functions of the objects, such as ds.filter(o => o.amount > 100), are functions that Spark can only call, not read, like the functions passed to an RDD: the optimizer described next can't rewrite them, and each row is converted to an object first. In Scala, a DataFrame is simply Dataset[Row], a Dataset of generic rows. Python has no Datasets: the Spark documentation notes that Python's dynamic typing already gives much of what they offer, so PySpark has DataFrames only.

A relational engine sits on top#

Since Spark 2.0 the recommended APIs are DataFrames (Datasets in Scala and Java) and SQL. Spark can optimize them because it sees the whole query as a tree of relational operators (scan, filter, join, aggregate) instead of opaque functions. Three parts of the engine matter from day one:

• The Catalyst optimizer rewrites the query before any of it runs. Because it sees operators rather than opaque code, it can rearrange them. Its rules move a filter down to the file scan, from where it is handed to the Parquet reader, so that Parquet skips every row group (a block of rows within a file) whose minimum and maximum values show that no row in it can match. It also makes the scan read only the columns the query uses. The planner that runs after it then chooses how to join: copy a small table to every executor (a broadcast join) or shuffle and sort both sides (a sort-merge join). None of this is possible for a filter written as a Python function over an RDD, which Spark can only call, not inspect.
• Tungsten stores rows as bytes rather than as objects. In plain JVM code, a row such as (42, 19.99, "DE") is a small object graph: the row, an array of its fields, and one object for each value, about 140 bytes in all, each piece tracked separately by the garbage collector, the part of the JVM that finds objects no longer in use and frees their memory. Tungsten writes the same row as 40 bytes into a page: one large block of memory (on the heap, a single Java array) that Spark fills with rows one after another. Spark sizes pages from each executor's memory, up to 64 MB, which is what our executors get with their 8 GB heaps: enough for over a million such rows. The garbage collector therefore sees one object instead of millions, and Spark sorts and shuffles rows by copying and comparing bytes, with no conversion to and from objects. Instead of passing every row from one operator to the next through a function call, Spark also generates Java code that fuses consecutive operators (filters, projections, aggregations) into one tight loop.
• Adaptive query execution (AQE) revises the plan while the query runs. At each shuffle, Spark knows the real size of every partition instead of an estimate, and AQE uses that to merge many small partitions into fewer tasks (at most about 64 MB each), to switch a join to a broadcast join when one side turned out small, and, in a join, to split a partition that is far larger than the rest, the skew case above.

We will see all three at work on the orders query below. A later post on how Spark SQL executes a query takes Catalyst and AQE apart and shows the generated code, and the post on storage, caching and memory takes Tungsten's binary rows apart.

I know these ideas sound abstract. Bear with me: after a short map of Spark's layers, we run one real query and watch these ideas at work.

The Layers#

The following table lists Spark's layers from the bottom up and when you touch each one directly.

Two consequences are worth stating early. First, the DataFrame API and SQL are the same thing to the engine: if customers is a DataFrame that reads the table crm.customers, then customers.groupBy("country").count() and SELECT country, count(*) AS count FROM crm.customers GROUP BY country produce the same optimized plan, so choosing between them is a matter of taste, not performance. Second, the plans that PySpark DataFrame code builds run in the JVM; only Python functions that Spark calls on the data, such as UDFs and RDD functions written in Python, move rows into Python workers, Python processes that the executors start, and that is why such functions cost so much more than built-in functions (more on this below).

How a Query Becomes Tasks#

The path from code to running tasks has two halves. The upper half is the SQL engine, which runs on the driver: it plans the query before any data moves and, with AQE, plans again after each shuffle. The lower half is Spark Core's two schedulers, also on the driver. They receive work as jobs: each action hands them at least one job, and they turn each job into stages and tasks and keep the executors busy. We follow our running example through both.

The code starts from spark, the SparkSession: the program's entry point to Spark, which a PySpark program creates first and a notebook usually creates for you.

from pyspark.sql import functions as F

orders = spark.read.table("sales.orders")        # about 100 GB of Parquet
customers = spark.read.table("crm.customers")    # 2 million customers, 54 MB

revenue = (orders.where(F.col("order_date") >= "2026-01-01")
           .join(customers.where(F.col("segment") == "B2B"), "customer_id")
           .groupBy("country")
           .agg(F.sum("amount").alias("revenue")))

revenue.write.mode("overwrite").saveAsTable("mart.revenue_by_country")

The numbers in the comments describe our tables. crm.customers holds 2 million customers in 54 MB of Parquet, in five columns (customer_id, name, segment, country and signup_date); a tenth of them, about 200,000, are business customers (segment is B2B), and they are spread over 30 countries, each with orders from 2026, so the result has 30 rows, one per country.

The same query with RDDs#

Written with RDDs, the same query starts from the rows of the two tables: .rdd turns a DataFrame into an RDD of its rows, as Python Row objects.

from datetime import date

b2b_rdd = (customers.rdd
           .filter(lambda c: c.segment == "B2B")
           .map(lambda c: (c.customer_id, c.country)))     # (customer_id, country)
revenue_rdd = (orders.rdd
               .filter(lambda o: o.order_date >= date(2026, 1, 1))
               .map(lambda o: (o.customer_id, o.amount))   # (customer_id, amount)
               .join(b2b_rdd)                              # (customer_id, (amount, country))
               .map(lambda kv: (kv[1][1], kv[1][0]))       # (country, amount)
               .reduceByKey(lambda a, b: a + b))           # (country, revenue)

revenue_rdd.collect() returns the same 30 rows as the DataFrame version, but Spark can do much less with this code. None of the relational engine's rewrites apply to the lambdas: Spark can't push the date filter into the Parquet reader or read only the columns the lambdas use, so it reads every column of every row. Each row becomes a Python object in a separate Python process, which the section on Python describes. And join shuffles both sides: RDD operations run as written, and PySpark's join always combines the two sides and shuffles them by key, so nothing can decide to copy the small side instead.

Planning on the driver#

Three components of the SQL engine take turns on the DataFrame version of the query, each with its own role:

• The analyzer checks the query against the catalog, which maps each table name to its schema, format and location: do sales.orders and crm.customers exist, and what are the types of their columns?
• The optimizer rewrites the plan of relational operators into a cheaper equivalent one, without yet deciding how each operator will run.
• The physical planner chooses an algorithm for each operator, such as which join strategy to use.

Our code already applies both filters to the tables before the join, so they sit directly on the two scans (a filter written after the join would be moved there by the optimizer), and the physical planner passes them into the Parquet reader. The order_date filter reaches the reader, which skips every row group whose statistics show only older dates. How much that saves depends on how the files were written. Suppose orders is filled by a nightly load that writes each day's orders into new files: the row groups then follow the dates, and the reader skips every row group from before 2026. We assume that most of our orders are from 2026, so this skips only an older minority; on a table with years of history, it would skip most of the row groups, although the reader still opens every file to read the statistics stored in it. The optimizer prunes every column except order_date, customer_id, amount, segment and country. It also adds isnotnull(customer_id) on both sides, because an inner join can never match a null key, and dropping those rows early saves shuffling them.

The physical planner then chooses how to join, and for that it estimates the size of each side. It knows that crm.customers takes 54 MB on disk, and that the join needs only two of its five columns, customer_id and country. It scales the 54 MB by the share of each row that these two columns take, using its own estimate of each column's width (36 of 80 bytes), and arrives at 24 MB. The B2B filter doesn't lower that estimate: without statistics on the segment column in the catalog, which Spark collects only when you run ANALYZE TABLE ... FOR COLUMNS (or FOR ALL COLUMNS) and uses only when its cost-based optimizer is turned on (spark.sql.cbo.enabled, off by default), the planner assumes that a filter keeps every row. Even those statistics wouldn't tell it that a tenth of the customers are B2B: for a text column, Spark records how many distinct values it holds but not how often each occurs, and assumes that each is equally common. Unless the code asks for a broadcast join, Spark copies a join's smaller side to every executor only when it estimates that side at 10 MB or less, the broadcast threshold, and 24 MB is above that. The planner therefore plans a sort-merge join: both sides are shuffled by customer_id, sorted, and merged. After the join comes a partial aggregation by country within each task, another shuffle, and the final aggregation feeding the write. A later post, query execution in Spark SQL, takes the analyzer, the optimizer and the physical planner apart.

Stages and adaptive query execution (AQE)#

As planned, before AQE changes anything, the query has four stages, cut at the three shuffles: scan and shuffle orders; scan and shuffle customers; join, partial aggregate and shuffle; and final aggregate and write.

AQE is on by default since Spark 3.2 (spark.sql.adaptive.enabled). With it, each shuffle stage first runs as a job of its own, and the rest of the plan is re-optimized when it finishes, now with real sizes instead of estimates. At runtime, the B2B customers turn out to be 6 MB of shuffle data: about 200,000 rows of 32 bytes each in Tungsten's binary format, counted before compression. A tenth of the planner's 24 MB would be 2.4 MB; the real figure is larger because the planner started from the compressed files, while the shuffle counts the rows uncompressed.

AQE replans the join as a broadcast hash join, which builds a hash table from those 6 MB of customers and copies it to every executor. The join also reads the orders shuffle output by map task instead of by partition: each of 200 join tasks reads the complete output of 4 of the 800 map tasks, and Spark prefers to run each task on one of the machines that wrote them.

AQE switches the join as soon as the customers' stage finishes, usually while the orders stage is still running. It doesn't stop that stage: it lets it finish and reuses its shuffle files, so the orders shuffle is still paid for. When you know that one side of a join is small, asking for a broadcast join in the code (a broadcast hint) avoids that shuffle from the start.

The final aggregation's input is small, too: each of the 200 join tasks sends at most one partial sum per country, so at most 30 rows, and the 200 default shuffle partitions hold at most 6,000 rows, about half a megabyte, most of the partitions nothing. AQE sees these sizes and coalesces the partitions into a single task. None of that is visible in explain() before the query runs; the SQL tab of the Spark UI, the web interface that the driver serves while the application runs, shows the final plan.

We chose the sizes of our tables to show several mechanisms with one query. With 2 million customers in 54 MB, a tenth of them B2B, the planner's estimate of the customers side is above the broadcast threshold, while the B2B customers that are actually shuffled stay well below it; so the same query plans a sort-merge join, and adaptive execution turns it into a broadcast join at runtime. Real tables rarely line up this neatly.

How the join and the aggregation run#

The plan names the operators; what each one does with its rows decides how much data moves and how much memory a task needs. Two of them do most of our query's work after the scan: the join and the aggregation.

The sort-merge join that the planner chose needs both inputs shuffled by customer_id, so that partition 17 of the orders and partition 17 of the B2B customers, for example, hold the same customers. Each join task sorts both of its partitions by customer_id and then reads them side by side, like merging two sorted lists: it moves forward on whichever side has the smaller customer_id, and when the two are equal, it outputs each order with the customer's country. An order whose customer isn't B2B finds no match and is dropped. The sort keeps as many rows in memory as it can and spills to disk when memory runs short, while the merge holds only the rows of the current customer_id.

The broadcast hash join that AQE switched to needs no sort. The driver collects the rows of the 200,000 B2B customers and builds a hash table keyed by customer_id that holds each customer's country; every executor receives one copy, which all of its join tasks share. Each join task then reads its orders one by one and looks up each customer_id in the table: a match produces a joined row, and an order whose customer isn't in the table is dropped. A lookup takes about the same time however large the table is, and the orders can arrive in any order and from any map task, which is why AQE can read their shuffle output by map task. The table is larger than the 6 MB of rows that crossed the shuffle, because it stores a link with each row for its lookups and reserves its memory in large blocks: the rows in it take about 8 MB, but each executor's copy takes under 80 MB.

The aggregation runs in two phases, on either side of the last shuffle. In the join stage, each task keeps a hash map from country to a running sum and adds the amount of every joined order to its country's entry; when its orders run out, it writes one row per country it has seen: at most 30, one for each country of our B2B customers, however many orders it read. Only these partial sums cross the shuffle, at most 30 rows from each of the 200 join tasks, about half a megabyte in all, instead of every joined order. The final aggregation, the single task that AQE left, adds up the partial sums of each country and produces the 30 rows of our result. This works because a sum can be built from partial sums; so can a count, a minimum and a maximum, and an average is kept as a sum and a count until the end.

In the join stage, the lookup and the partial sum don't run as separate steps: the generated code we met in the mental model fuses both into the loop that reads the orders, so each order is read, looked up and added to its country's sum before the next one is read.

The physical operators get a post of their own. In short, it covers how joins, aggregations, sorts, window functions and exchanges work inside, and how much memory each needs.

Choosing the number of partitions#

Our query has used the defaults so far, and each of them can be changed. spark.sql.files.maxPartitionBytes (128 MB) sets how much of the input goes into one partition: at 256 MB, the 100 GB of orders become 400 partitions, and the scan runs in 10 waves instead of 20, with twice as much data per task. spark.sql.shuffle.partitions (200) sets how many partitions a shuffle hashes the rows into. AQE then merges neighboring small partitions. By default it aims for about one partition per core, 40 on our cluster, but for no less than 1 MB (spark.sql.adaptive.coalescePartitions.minPartitionSize) and no more than 64 MB (spark.sql.adaptive.advisoryPartitionSizeInBytes) per partition. Our 200 aggregation partitions hold half a megabyte in all, less than the 1 MB minimum, so they became one. In code, df.repartition(400, "customer_id") shuffles a DataFrame into 400 partitions by a column, and df.coalesce(10) merges its partitions into 10 without a shuffle.

Scheduling: from stages to tasks#

Below the SQL engine, every query is run by the same two schedulers on the driver. Spark's source calls them the high-level and the low-level scheduling layers, and they split the work at one line: the DAGScheduler thinks in stages, the TaskScheduler in tasks.

For each job, the DAGScheduler walks the plan's underlying RDD graph backward from the result, cuts it into stages at every shuffle boundary, and submits a stage once all its parents' shuffle output exists. For our query without AQE, it would submit the two scan stages first, since they have no parents; with AQE, each scan stage runs as a job of its own, and the DAGScheduler handles each of these jobs the same way. Each submitted stage becomes a TaskSet, one task per partition, which the TaskScheduler places on free executor cores. If a task has a preferred location (for example, an executor that has its input cached), the TaskScheduler waits a few seconds for a core there before placing it anywhere. On the executor, a thread deserializes the task, runs the pipelined operators over its partition, writes shuffle output or returns a result, and reports back to the driver with the task's metrics. When a stage's last task has finished, the DAGScheduler submits the stages that were waiting for it.

Failures show the split best. A task that hits a corrupt file or a bug in a UDF fails while its input is still there: the TaskScheduler runs it again, and only after four failed attempts does it give up; the stage fails, and with it the job. Suppose instead that a machine dies after the scan stages have finished. The scan tasks that ran on it wrote their shuffle files to its local disks, so those files are gone, and the join stage's tasks can't fetch them. Retrying the join tasks can't help, because their input no longer exists. The DAGScheduler steps in: it re-runs the scan tasks whose output was lost, then the join stage. This repair has a limit too. Each time the join stage loses its input this way, its attempt counts as failed, and after four failed attempts in a row (spark.stage.maxConsecutiveAttempts), Spark stops and fails the job, so that a job whose input keeps disappearing ends with an error instead of retrying forever.

What losing only an executor costs, say because its JVM crashed, depends on the setup. Many YARN and standalone clusters run an external shuffle service: a service on each machine, outside the executors, that keeps serving the shuffle files of executors that are gone. On YARN it runs inside the NodeManager, the YARN process on each machine that starts executors there; on a standalone cluster, inside the worker, the process that does the same. Spark has none for Kubernetes. With the service, the finished map tasks of a lost executor need not re-run; only the tasks the executor was running are retried. Without one (Spark's default), each executor serves its own files, and they disappear with it.

The next post, Jobs, stages and tasks, follows this path step by step: the threads behind both schedulers, what happens when the driver can't keep up, and what happens when a task, an executor or a shuffle file is lost.

Writing the result#

All of this work started with saveAsTable, and the write is part of the same query: the tasks of the final stage write the output. Each one finishes the aggregation for its partition and writes its rows as one Parquet file, compressed with Snappy by default. A task whose partition is empty writes nothing, except the task for partition 0, which always writes a file so that the table's schema is recorded. After AQE merged the 200 shuffle partitions into one, our query writes one file. Without AQE, it would write one small file for every non-empty partition, at most one per country, and one more for partition 0 if that partition is empty: up to 31 in all. For a table like ours, whose files all sit in one directory, the number of tasks in the last stage sets the most files the write can produce: by default, one per task.

The tasks don't write into the table's directory directly. If they did, a reader could see a half-written table, or two copies of a file whenever a task ran twice. Spark's default commit procedure, which comes from Hadoop, works in three steps instead:

1. Each attempt of a task writes its file into a directory of its own, under _temporary/ inside the table's directory. This directory is on the same storage as the table (HDFS or a cloud store), not on the executor's local disk. No other task writes into it, and readers don't look into it: Spark, like most tools that read tables, skips files and directories whose names start with an underscore. Because the directory belongs to one attempt of one task, a retried task writes into a different directory than the attempt that failed.
2. When the task has finished writing, it asks the driver for permission to commit. The driver grants this to only one attempt of each task, so if a task ran twice, only one attempt commits. The task then renames its attempt's directory, still under _temporary/, to one named after the task alone.
3. When every task has committed, the driver commits the job. It renames every file from those directories into the table's directory, deletes _temporary, writes an empty _SUCCESS file, and only then creates the table's entry in the catalog.

Every step that publishes data is a rename, and what a rename costs depends on the storage. On HDFS, on a local disk, and on cloud storage with a hierarchical namespace (Azure Data Lake Storage Gen2 with the hierarchical namespace enabled, or a Google Cloud Storage bucket created with one), directories are real. Renaming a file or a directory changes one entry in the directory tree: no data moves, however large the files, and the rename is atomic. A job commit that moves many files is still a series of renames, though, so a reader that lists the directory in the meantime can see some of the files and not others.

Flat object stores have no real directories: Amazon S3's general purpose buckets, Azure Blob Storage without the hierarchical namespace, and Google Cloud Storage buckets without one keep each file as an object whose name merely contains slashes. There, renaming a directory means copying every object under that name prefix to its new name and then deleting the originals. It takes longer the more files and bytes there are, and other readers can see it half done. (S3 Express One Zone directory buckets gained an atomic rename in 2025, but only for single objects.) Writing to a flat object store therefore calls for commit procedures that avoid renames altogether. Hadoop's committers for S3, for example, start an upload of each file to its final name but complete the uploads only when the job commits, so until then readers see nothing.

A reader that goes through the catalog sees our table only once all its files are in place. Then saveAsTable returns, and our query is done.

After the commit, a table written by two tasks looks like this:

revenue_by_country/
  _SUCCESS
  part-00000-<job id>-c000.snappy.parquet
  part-00001-<job id>-c000.snappy.parquet

Our query writes only part-00000; we added part-00001 for illustration, to show how the files of several tasks are named. The number after part- is the task's partition, and <job id> stands for a random identifier (a UUID) of the write job. The number after c counts the files that one task writes; it goes past 000 only when a task splits its output, for example because spark.sql.files.maxRecordsPerFile caps the rows per file.

Our code asked for mode("overwrite"). The save mode decides what saveAsTable does when the table already exists; when it doesn't, every mode creates it.

append suits a table that grows, such as a daily load of new orders; overwrite suits a table that is rebuilt from scratch each time, like ours. Neither fits a table stored with one directory per day, such as order_date=2026-10-10/, in which a job rewrites only the days that received late orders: append would add a second copy of those days, and overwrite would drop every other day.

For that, there is insertInto, which writes into an existing table and keeps its definition. With mode("overwrite") and spark.sql.sources.partitionOverwriteMode set to dynamic, it replaces only the day directories that the query writes rows into and leaves the others alone; with the default, static, it first deletes every day of the table. Unlike saveAsTable, insertInto matches columns by position, not by name.

The read side mirrors the write. Spark asked the catalog where sales.orders lives, listed its files, and cut them into the 128 MB splits that became our 800 input partitions. With a table format such as Iceberg, the list of files comes from the table's metadata instead of a directory listing, and a commit is one atomic update of that metadata.

How a table is laid out decides how much of that listing a query can skip. With one directory per day, like the table in the write modes above, a filter on order_date removes whole directories while Spark plans the query, before it opens a single file; this is partition pruning (a table's directories are also called its partitions, which are not the partitions that tasks process), and it works on the directory names alone. Our filter would skip every day before 2026 that way. Choosing the column matters: one directory per customer_id would split our 100 GB into up to 2 million directories, about 50 KB each on average, and millions of small files cost more to list and open than to read.

A layout can also save a shuffle. A table written with bucketBy(200, "customer_id") is bucketed: Spark hashes each customer_id to a number from 0 to 199, the same way a shuffle into 200 partitions would, and each file holds rows with a single number. A join of two tables bucketed the same way on the same key then needs no shuffle, because the rows are already split by the key.

Reading and writing files get a post of their own. In short, it covers how Spark finds and prunes the files it reads, why many small files hurt on both sides, static and dynamic overwrite, and the commit procedures that write to flat object stores such as S3 without renaming anything.

Beyond One Query#

Our query showed the path that every DataFrame and SQL query takes. Around that path sit the topics that decide whether the same query runs well on a real cluster: how an executor uses its memory, how Spark keeps data around for reuse, what changes when Python code has to see the rows, how to find out why a job is slow, how the same query runs as a stream, how an application gets onto a cluster in the first place, and how Spark Connect separates your program from the driver. Each gets a tour here, and later posts of the series take them apart.

Executor memory#

An executor is a JVM process on one of the cluster's machines, but it doesn't get to use that machine's memory freely. When the driver asks for an executor, it asks for a fixed amount of memory and a number of cores, and the cluster manager reserves at least that on some machine (YARN's default scheduler rounds every request up to whole gigabytes). On YARN this reservation is called a container; on Kubernetes, the executor runs in a pod, which plays the same role. The reservation is enforced: if the executor's process uses more memory than its container allows, it is killed (on YARN by the NodeManager; on Kubernetes by the kernel's out-of-memory killer). The driver then loses the executor together with its running tasks and its cached data. Whether the shuffle files it wrote are lost too depends on the setup we met in the failure example: an external shuffle service on the machine keeps serving them, and without one they are gone with the executor. Unlike YARN and Kubernetes, Spark's standalone master reserves only the heap and enforces no limit.

By default, the memory Spark asks YARN or Kubernetes for has two parts. The first is the JVM heap: the memory where the JVM keeps Java objects, and where its garbage collector finds the objects that are no longer used and frees their space. spark.executor.memory sets the heap's size. The second part is the memory overhead: everything the process uses outside the heap, such as the JVM's own code and metadata, the stacks of its threads and the buffers of its network connections. For PySpark it also covers the Python worker processes we meet in the next sections, which are separate processes but run inside the same container. By default, the overhead is 10 percent of the heap, but at least 384 MB.

For PySpark, that can be too little, because the Python workers count against the overhead, and a job that runs heavy UDFs can outgrow 10 percent. Spark's Kubernetes support anticipates this: since it gained Python support in Spark 2.4, it gives PySpark applications a default of 40 percent in cluster mode, because, as its documentation puts it, such jobs "commonly fail with 'Memory Overhead Exceeded' errors". Nothing about YARN makes Python need less. YARN support simply kept the general default, so a PySpark job on YARN that needs more room outside the heap has to raise spark.executor.memoryOverhead itself.

The overhead comes on top of the heap, not out of it. Our executors run with spark.executor.memory=8g, so each JVM gets a full 8 GB heap, and on YARN, where our job runs, Spark asks the cluster manager for a container of 8 GB plus about 0.8 GB of overhead: about 8.8 GB in all, which YARN rounds up to a 9 GB container.

Spark divides the heap further, into a part it sets aside for itself, a part it manages for running tasks and cached data, and a part it doesn't manage. It first sets 300 MB aside for its own internal objects. Of the rest, 60 percent (about 4.6 GB here) is unified memory, which Spark manages itself and uses for two purposes:

• Execution memory is working space for tasks while they compute: the hash table an aggregation builds, the buffer a sort fills, the records a shuffle collects before writing them to disk. It is needed only while a task runs.
• Storage memory holds data that Spark keeps for later: cached DataFrames, which the next section explains, and broadcast tables such as the hash table of customers that AQE copied to every executor for our join.

The remaining 40 percent (about 3.1 GB) is user memory, which Spark doesn't manage. It holds three things. The first is the objects that JVM code you write creates: the variables and collections in a Scala or Java UDF, or a lookup map that an RDD function such as mapPartitions builds. (A PySpark UDF's objects live in the Python worker process instead, so they use the overhead, not the heap.) The second is Spark's own internal metadata. The third is a safety margin. Spark keeps count of the memory its operators use, but part of that count is an estimate, and some objects are never counted: the Parquet reader, for example, decodes 4,096 rows at a time into plain Java arrays outside Spark's accounting. An unusually large record, such as a row with a 500 MB string, makes the real usage exceed the count. The unmanaged 40 percent leaves room for that difference, so that it doesn't end in an OutOfMemoryError.

Execution and storage don't get fixed halves of the unified memory. The boundary between them starts in the middle (spark.memory.storageFraction, 0.5), and either side may grow into space the other isn't using. A cache can fill memory that no task needs at the moment. Running tasks, in turn, can take storage space that holds no cached data, so execution can grow past its half, up to all of the unified memory when nothing is cached.

The two sides are not equal when they collide, though. If running tasks need memory that cached data has taken from them, Spark evicts only as many cached blocks as the tasks need, and never below storage's half; evicted blocks are moved to local disk or dropped, depending on how they were cached. Cached data inside storage's half is never evicted for execution. The reverse never happens: when tasks have grown into storage's half, a cache can't take that memory back and simply caches less, until the tasks finish or free it. Spark's source gives the reason plainly: taking memory away from a running task would be too complex to implement.

Within execution memory, the running tasks share. An executor with 4 cores runs up to 4 tasks at a time, and Spark limits each task to at most a quarter of the execution memory, so that one task can't starve the others. In general, with N tasks running, each task may hold at most 1/N.

The limit follows the number of running tasks, and that number changes. Suppose only two tasks were running at first: each could take up to half of the execution memory, and together they may have taken all of it. Then two more tasks start, and the limit drops to a quarter. The two earlier tasks keep what they hold, so the newcomers find little or nothing free. For this situation, Spark guarantees each task an eighth, 1/(2N) in general: a task that holds less than that and asks for more doesn't spill, but waits until other tasks free memory. A task between an eighth and a quarter gets whatever is free, and spills if nothing is. The early tasks free memory soon, because a task above the new limit of a quarter gets nothing more when it asks, and has to spill or finish.

A task that reaches its limit and still needs memory usually doesn't fail. The operator that asked, a sort or an aggregation, spills: it writes what it holds to local disk, continues with an empty buffer, and merges the files from disk when it finishes. Spilling makes a task slower, because it writes and reads its data an extra time, but the task completes.

This is why an out-of-memory error is rarely about execution memory, whose operators can almost always spill (a task fails with Unable to acquire ... bytes of memory only when even spilling can't free enough for one request). It usually points to something that can't spill, such as a single enormous record or a broadcast table too large to build, or to memory Spark doesn't manage: user memory, which holds your code's objects and what Spark doesn't count, and the overhead, which YARN or Kubernetes enforces.

Our query stays far from these limits. Its scan tasks ask for no execution memory at all. Besides the Parquet reader's arrays, each task keeps a few small buffers for each shuffle partition, about 100 KB, through which it writes its rows to local disk, compressing them as it goes (Spark compresses shuffle files by default): about 20 MB for 200 partitions. The arrays and the buffers are user memory. (Spark writes a shuffle this way when it has at most 200 partitions, a default threshold; with more, it first collects the rows in execution memory and sorts them by partition.) The hash table of B2B customers that AQE broadcasts sits in storage memory: one copy of under 80 MB on each executor, which all four of its tasks read, less than 2 percent of the 4.6 GB. The partial aggregation keeps one running sum per country, 30 per task, in a hash map in execution memory. The map reserves memory a page at a time, so even 30 sums take a whole 64 MB page in each task: about 260 MB for an executor's four tasks, under 6 percent of the 4.6 GB. Memory becomes the limit for operators that must hold much of their input at once, such as a large sort or an aggregation over millions of keys, and for caches, which the next section covers.

Caching#

The mental model showed that an action doesn't keep the rows it computed: a DataFrame used by two actions is computed twice. Caching avoids the second computation. It keeps a DataFrame's partitions on the executors after the first action has computed them, so that later actions read them from there instead of reading and computing them again. It pays off when the same intermediate result feeds several actions, as in two reports that start from the same B2B customers:

b2b = customers.where(F.col("segment") == "B2B").cache()

b2b.groupBy("country").count().show()          # 1st action: computes b2b and caches it
b2b.where(F.col("country") == "DE").count()    # 2nd action: reads the cached partitions

cache() does nothing by itself: like a transformation, it only marks the DataFrame. The first action that computes b2b stores each of its partitions on the executor that computed it, in storage memory, as compressed batches of columns. The second action finds the partitions there and skips the scan of crm.customers and the filter. The Storage tab of the Spark UI lists every cached DataFrame with how much of it sits in memory and how much on disk, and b2b.unpersist() removes it when it is no longer needed.

Where the cached partitions go, and what happens when storage memory is full, is set by the storage level. cache() uses the default; persist() takes the level as an argument: .persist(StorageLevel.DISK_ONLY) in place of .cache(), after from pyspark import StorageLevel, would keep b2b on local disk only. A DataFrame that is already cached keeps its level until unpersist(). When storage memory runs out, Spark evicts the partitions that were used least recently, though never another partition of the DataFrame it is caching at that moment.

Executor memory and caching get a post of their own. In short, it follows memory from the container's limit down to the pages a sort writes into, explains every storage level and how cached blocks are evicted, and shows when caching backfires.

Python and the JVM#

Our query is PySpark code, yet the executors never ran any Python. A PySpark program doesn't process data in Python; it describes a query. In a PySpark application, the driver side consists of two processes: your Python program, and the driver JVM, which the program controls through Py4J, a library that forwards Python method calls to objects in the JVM over a local socket. When the program calls orders.where(...), nothing is filtered in Python: Py4J asks the JVM to add a filter to the plan. Every call we made, from spark.read.table to saveAsTable, built or ran a plan in the JVM, and the executors ran that plan as JVM code, exactly as if we had written the query in Scala.

That holds as long as the query uses Spark's built-in functions: the column expressions in pyspark.sql.functions, such as F.sum, F.upper or F.to_date, several hundred in all, which are implemented in the JVM. Sooner or later, a pipeline needs logic that no built-in function provides: a calculation from a Python library, a company's own parsing rules, a call to a trained model. For that, PySpark lets you wrap a Python function as a user-defined function (UDF), which Spark calls for every row. Suppose the clean crm.customers we have been using is built from a raw table, crm.customers_raw, where country names are stored as people typed them ("Germany", "germany "), and the build has to turn them into ISO codes ("DE"). The Python library pycountry knows the mapping, and Spark has no built-in function for it:

import pycountry

@F.udf("string")
def to_iso_code(name):
    try:
        return pycountry.countries.lookup(name.strip()).alpha_2
    except (AttributeError, LookupError):    # a null, or a name pycountry doesn't know
        return None

(spark.read.table("crm.customers_raw")
    .withColumn("country", to_iso_code("country"))
    .write.mode("overwrite").saveAsTable("crm.customers"))

The executors' JVMs can't run this function, so Spark has to send the rows to Python and back. When it plans the query, it gives the UDF a step of its own in the physical plan, which explain() shows as ArrowEvalPython. On each executor, that step works like this:

1. The task collects the values the UDF needs (here, country) in batches of up to 10,000 rows and converts each batch to Apache Arrow, a columnar data format that both the JVM and Python can read.
2. It sends the batch over a local socket to a Python worker: a Python process that the executor starts next to itself and reuses for later tasks.
3. The worker turns the batch into Python values and calls to_iso_code once for each row.
4. It sends the results back as an Arrow batch, and the task continues in the JVM with the new column.

Arrow has been the default for Python UDFs since Spark 4.2; before that, rows traveled in Python's own serialization format, pickle. Arrow speeds up only the transfer, though. Step 3 really is a loop over rows: for a full batch, the worker calls to_iso_code 10,000 times, and each call goes through the Python interpreter, which executes the function's bytecode one instruction at a time, checks the type of every value it touches, and creates a new Python object for every result.

Compare that with how the JVM runs the rest of our query. Whole-stage code generation, the part of Tungsten we met in the mental model, fuses consecutive operators (a filter, a projection) into one loop of generated Java code, and the JVM compiles that loop into machine code. Each row then flows through all those operators as compiled instructions, with no function call from one operator to the next. A Python UDF interrupts this: Spark can't generate code for a step that runs in another process, so the generated loop ends before ArrowEvalPython and a new one starts after it, with the round trip to Python in between.

Each step of that round trip costs time that a built-in function doesn't: converting values to Arrow and to Python objects, copying them between two processes, running the interpreter once per row, and converting the results back. The Python workers also need memory, and it comes out of the container's overhead, not the heap. The first rule follows directly: use a built-in function whenever one exists. If the raw column held country codes typed as " de", for example, trimming and upper-casing them is F.upper(F.trim("country")), which runs inside the generated loop, with no round trip to Python.

When a UDF is unavoidable, a pandas UDF removes most of the per-row cost. Spark hands it each Arrow batch as pandas Series (whole columns of values) and expects a Series of results back, so Python calls the function once per batch instead of once per row. That pays off when the function itself works on whole columns, as pandas, NumPy and machine-learning libraries do. Scoring customers with a trained scikit-learn model is the typical case, here the probability that a customer orders again within a month:

import pandas as pd

# model: a trained scikit-learn classifier, loaded on the driver and shipped with the function
@F.pandas_udf("double")
def reorder_probability(order_count: pd.Series, total_revenue: pd.Series) -> pd.Series:
    features = pd.DataFrame({"orders": order_count, "revenue": total_revenue})
    return pd.Series(model.predict_proba(features)[:, 1])

# customer_stats: one row per customer, with its number of orders and its revenue
scored = customer_stats.withColumn("reorder_probability", reorder_probability("orders", "revenue"))

Here predict_proba receives up to 10,000 customers in one call. Inside it, NumPy keeps the values in contiguous arrays of plain numbers and runs the arithmetic in loops that were written in C and compiled to machine code ahead of time. The interpreter executes a handful of calls per batch, and the per-row work runs as compiled code. It is the same idea as whole-stage code generation in the JVM: keep the loop over rows out of the slow layer.

Arrow also speeds up moving a result from the JVM into the driver's Python process. toPandas() collects a whole DataFrame into the driver as a pandas DataFrame, for plotting or for a library that works only locally. With Arrow, the result arrives as column batches rather than row by row, but it still has to fit in the driver's memory, so toPandas() suits results, not tables.

PySpark's processes, the cost of each kind of UDF, Arrow and Spark Connect, which we meet after the section on running on a cluster, get a post of their own. In short, it explains what each crossing between the JVM and Python costs and which kind of UDF to choose when no built-in function will do.

When the batch query is slow#

Suppose we decide to report revenue by the country each order is shipped to, instead of the customer's country. The orders carry it in a column ship_country, typed in by customers as freely as the names in crm.customers_raw, and we assume it holds a few hundred different spellings. We add a column with to_iso_code from the previous section and group by it instead of country:

orders = orders.withColumn("ship_iso", to_iso_code("ship_country"))
revenue = (orders.where(F.col("order_date") >= "2026-01-01")
           .join(customers.where(F.col("segment") == "B2B"), "customer_id")
           .groupBy("ship_iso")
           .agg(F.sum("amount").alias("revenue")))

The first run with this change takes forty minutes instead of two. Where the time went is shown in the Spark UI, which the driver serves on port 4040 while the application runs. After the application ends, Spark's history server, a separate web server, can rebuild the UI from the application's event log, a file that records what the UI shows, if the application wrote one (spark.eventLog.enabled, off by default). The path through the UI is the same each time. The SQL tab shows the query's plan with the number of rows each step produced. The Jobs and Stages tabs show which stage took the time. The page of that stage shows why.

The stage page's summary metrics table is the place to start. For each measurement Spark takes per task (how long it ran, how much data it read, how much it spilled, how long it spent in garbage collection), the table shows the minimum, the 25th percentile, the median, the 75th percentile and the maximum over the stage's completed tasks. A healthy stage has a maximum not far from its median. Four patterns explain most slow stages:

Read the median and the maximum side by side; most of a diagnosis is in that gap.

Skew is the pattern that wastes the most, because a stage can't finish before its slowest task: one oversized partition keeps every other core waiting. The stage page's event timeline, which draws each task as a bar over time, shows it at a glance. AQE handles the common case in joins on its own: it splits a partition that is more than five times the median size and larger than 256 MB into several tasks. For the garbage collector, the Executors tab helps too: it marks in red the executors whose garbage collection took more than 10 percent of their task time.

Our slow query matches none of these patterns. The Stages tab puts almost all of the forty minutes in the stage that scans orders. On its page, the median task takes about as long as the slowest, nothing spilled, and garbage collection takes little time: every task got slower by about the same amount, so the cause lies in the work each task does, not in how the data is spread. The SQL tab names that work. Between the scan and the shuffle, the plan now has an ArrowEvalPython step, and its metrics include time to execute Python code: the interpreter's per-row loop from the section on Python, run once for every order from 2026. But ship_country holds only those few hundred spellings, so Python needs to see each spelling once, not each order. Instead of the withColumn line:

spellings = orders.where(F.col("order_date") >= "2026-01-01").select("ship_country").distinct()
codes = spellings.withColumn("ship_iso", to_iso_code("ship_country"))
orders = orders.join(F.broadcast(codes), "ship_country", "left")

This adds a second pass over the ship_country column of the 2026 orders and a distinct that keeps a few hundred rows, both in the JVM, while to_iso_code runs a few hundred times instead of once per order. F.broadcast asks for a broadcast join from the start: the planner can't know that distinct leaves only a few hundred rows, so without it, every order would be shuffled by ship_country even if AQE switched the join later, as happened to the orders in our join with the customers.

Tuning and troubleshooting closes the series. In short, it covers sizing executors, reading every tab of the UI, each kind of out-of-memory error, and a playbook of worked diagnoses.

Streaming#

So far orders was a table that a nightly load fills. Suppose instead that every order is published as an event to Kafka, a system that stores streams of events in named topics, here orders. A topic is split into partitions, so that many applications can write and read it in parallel. These are Kafka's partitions, not Spark's, although Spark reads each one with a task of its own, as we'll see. Each partition is an ordered, append-only log kept on one of Kafka's servers, its brokers (and usually copied to others, so that it survives the loss of one): new events are added at the end, and nothing is changed in place. The application that publishes an order chooses the partition, usually by hashing a key such as the customer id. Within a partition, every event has a position, its offset, which only grows, so a reader knows exactly how far it has read by remembering one offset per partition.

The business now wants revenue per country for every hour, as soon as the hour is over. Structured Streaming, Spark's engine for streams, lets us write that as almost the same query:

orders = (spark.readStream.format("kafka")
          .option("kafka.bootstrap.servers", "broker:9092")
          .option("subscribe", "orders")
          .load()
          .select(F.from_json(F.col("value").cast("string"),
                              "customer_id BIGINT, amount DECIMAL(12,2), order_ts TIMESTAMP")
                  .alias("o"))
          .select("o.*"))

revenue = (orders.withWatermark("order_ts", "10 minutes")
           .join(customers.where(F.col("segment") == "B2B"), "customer_id")
           .groupBy(F.window("order_ts", "1 hour"), "country")
           .agg(F.sum("amount").alias("revenue")))

query = (revenue.writeStream
         .outputMode("append")
         .option("checkpointLocation", "/checkpoints/revenue_hourly")
         .trigger(processingTime="1 minute")
         .toTable("mart.revenue_by_country_hourly"))

Apart from the date filter, which a stream of new orders doesn't need, four things changed compared with the batch query. readStream reads the orders topic as a table that never ends: its rows are the events that have arrived since the query started and the ones that will arrive. It reads through the Kafka source, a separate package (spark-sql-kafka-0-10) that has to be added to the application when it is launched. Each event carries its order as JSON in its value column, which .cast("string") turns from bytes into text and from_json into columns, including order_ts, the time the order was placed. F.window("order_ts", "1 hour") assigns each order to the hour it was placed in, so the grouping is now by hour and country. withWatermark tells Spark how late an order may still arrive. And writeStream describes the output: outputMode says which result rows to write, checkpointLocation names a directory on reliable storage where the query records its progress (a different kind of checkpoint from the RDD checkpoints above), and trigger says how often to process new data. toTable at the end starts the streaming query, which writes into mart.revenue_by_country_hourly and keeps running instead of finishing like our batch query. customers is the same static table as before; how a stream joins a static table is shown below. The rest of this section takes these parts one at a time: how the query runs, when a result is final, what it writes, where it keeps what it remembers, and how it survives a crash.

Micro-batches and triggers#

Spark runs a streaming query as a sequence of small batch jobs, called micro-batches. Each micro-batch takes the events that arrived since the previous one, that is, the Kafka offsets between where the last batch stopped and where each partition of the log ends now, and runs the planned query on just those rows, with the same kind of plan, stages, shuffles and tasks as our batch query, on a much smaller input. One difference: adaptive query execution is off for streaming queries that keep state from one batch to the next, such as ours, so each batch runs its plan as planned.

Reading from Kafka works like this. At the start of each micro-batch, the driver asks Kafka for the latest offset of every partition of orders. Together with where the previous batch stopped, that gives one offset range per partition, for example offsets 1,200 to 1,450 of partition 0. Spark plans one task per range, so a topic with 3 partitions is read by 3 tasks in parallel; the source's minPartitions option splits large ranges into more tasks when a few partitions are too much work for one task each. Each task connects to the broker that holds its partition and reads exactly its range. Spark tries to run the task for a given partition on the same executor in every batch, so that the executor can keep its connection to Kafka open. The offsets themselves are recorded by Spark, not by Kafka; the section on checkpointing below shows where.

The join with customers happens inside each micro-batch. The new orders of a batch are joined with all of the B2B customers, and only the joined rows go on to the hourly sums. Nothing from the join is remembered between batches: customers is read again in every batch, and an order that has been joined is done. This is the difference from joining two streams, where each side has to keep its rows from one batch to the next until the matching rows of the other side arrive.

The trigger decides when the next micro-batch starts:

• Without a trigger, a new batch starts as soon as the previous one has finished. This gives the lowest latency the micro-batch engine can offer, and keeps the cluster busy.
• With trigger(processingTime="1 minute"), as in our query, batches start on the minute. If a batch takes longer than a minute, the next one starts immediately after it, and Spark logs the warning "Current batch is falling behind".
• With trigger(availableNow=True), the query processes everything that is available when it starts, possibly in several batches, and then stops. That turns a streaming query into an incremental batch job: run it once an hour from a job scheduler outside Spark, such as cron, and each run picks up exactly where the previous one ended.

In each case, a batch runs only if there is something to do. Without new events, Spark skips the batch, with one exception that the section on watermarks explains: a stateful query may run one more batch without new data, when the data it has just read has moved its watermark, the order_ts up to which results can be final.

Two other modes keep tasks running instead of starting short batches. The older continuous processing mode, still marked experimental, keeps long-running tasks reading events one by one for millisecond latency, but supports only simple queries. A newer mode, the real-time mode of Spark 4.1, also keeps tasks running and processes events as they arrive. It is set with trigger(realTime="5 minutes"), available in PySpark since 4.2; the 5 minutes are the length of each long-running batch, not a latency. In open-source Spark, real-time mode supports only queries of a single stage without state, so our hourly aggregation can't use it.

Watermarks#

An hourly sum can't be finished from one minute of orders, so the aggregation keeps the sums so far, per hour and country, from one batch to the next. Nothing in our code asks for this explicitly: Spark keeps state because the query aggregates a stream, and the groupBy on F.window("order_ts", "1 hour") and country decides what it keeps, one sum per hour and country. Everything a streaming query must remember between batches is its state: the sums so far of an aggregation, the keys a deduplication has already seen, the rows of one stream waiting for their partner in a stream-stream join.

Streaming systems distinguish two kinds of time. Event time is when something happened, here when the order was placed (order_ts). Processing time is when the system sees it. The two differ, and not by a fixed amount: events are delayed by networks, queues and devices that were offline, so they arrive out of order. A watermark is a streaming system's measure of progress in event time: a timestamp W with which the system declares that it no longer expects events whose event time is earlier than W. A watermark only moves forward, and it is what lets a system decide that the results for a period of event time are complete. Events that arrive behind the watermark are late.

For our query, the watermark answers a practical question. State has to be removed at some point, or it grows forever. For our hourly sums, the natural moment is when an hour is complete. But events don't arrive in the order in which they happened: a phone that was offline sends its orders minutes later. So when is the 10:00 hour complete?

Spark's watermark answers that, and it moves only between batches. While a batch runs, Spark notes the latest order_ts among the orders it reads. When the batch has finished, the driver subtracts the 10 minutes we allowed for late events from the latest order_ts seen so far; the result is the new watermark, the point in event time before which Spark expects no more orders. The batch that the trigger starts next works with this watermark: with our one-minute trigger, the batch on the next full minute, or the one that starts right away if the last batch ran past it. That batch first adds its own orders to the sums, then writes every hour that ends at or before the watermark as final and removes it from the state.

The chart below shows both times at once. Each order is drawn as a dot: its horizontal position is the time it was placed, its order_ts, and its height is the time a batch read it. Suppose orders arrive continuously, hundreds a minute, each some seconds to a couple of minutes after it was placed: they then lie just above the dashed diagonal, which marks an order read the very moment it was placed.

The staircase is the watermark as Spark actually computes it. Each step is where one batch ends and the next begins. It changes only between batches, and only upward: after each batch, Spark takes the latest order_ts read so far and subtracts 10 minutes, so the watermark trails the orders by a little over 10 minutes. Each circle marks the newest order one batch read; the watermark that batch sets lies 10 minutes earlier in event time, where its dashed arrow ends. The steps are uneven, because each depends on when a batch starts and on how recent the newest order it reads is; when a batch runs long, as the one from 11:03 to 11:05:30 does here, the watermark waits and then jumps. It passes 11:00 after the batch at 11:11; the next paragraph zooms in on that step. A late order never moves the watermark back, because only the latest order_ts counts.

Zoom in on the step that closes the 10:00 hour. The newest order the batch at 11:11 reads was placed at 11:10:33. When that batch ends, the driver sets the watermark to 11:00:33, and the batch the trigger starts at 11:12 is the first to work with it: it adds its orders to the sums, then writes the 10:00 hour as final and drops it from the state. Until then, the sums for the 10:00 hour live only in the state, which Spark saves to the checkpoint after every batch. In append mode, which our query uses, they reach the table once, when final; update mode, described below, writes an hour's new sum after every batch that changes it.

That leaves a gap when orders stop. The trigger starts a batch only when there is something new to read, so after the last order of the night, the batch that would write a finished hour would never come. Spark closes the gap: when a batch has moved the watermark, Spark runs the next batch even if no orders have arrived. This batch without data reads nothing; its only work is to write the hours the new watermark has closed. It can't close the hour of the night's last orders, though: only a new order moves the watermark, so that hour stays in the state until orders arrive again. It is the exception mentioned in the section on triggers. Our query, with hundreds of orders a minute, rarely needs one.

What happens to a late order depends on whether its hour is still open, not only on how late it is. Order A, placed at 10:50 and read at 11:09, is already behind the watermark of that moment (about 10:58), but the 10:00 hour is still open, so the order is added to its sum. Order B, also placed at 10:50 but held up until 11:25, finds the hour final, written and removed from the state, and is dropped. The query's progress report (the metrics Spark publishes after every batch, which query.lastProgress returns) shows such drops as rows dropped by the watermark. It counts rows after the batch's orders have been summed, one per hour and country, not orders: fifty late orders for the 10:00 hour in Germany that arrive in the same batch count as one. Spark guarantees only one direction: an order less than 10 minutes behind the latest order seen is never dropped. Older orders are counted as long as their hour is still open.

This holds in the append and update output modes described next. In complete mode, the watermark has no effect: Spark keeps every hour in the state forever and adds late orders whenever they arrive, because it writes the whole result table after every batch anyway.

Output modes#

A streaming aggregation produces a result table that keeps changing: after every batch, some rows are new or updated, and some have become final. The output mode decides which of these rows each batch writes to the sink, the destination of the query's results:

• Complete writes the whole result table after every batch, every hour and country seen so far. It is allowed only for aggregations, and it keeps all state forever, because the next batch must write all rows again; the watermark neither removes state nor drops late orders. It suits small results, such as a dashboard's totals.
• Update writes only the rows that changed in this batch. A dashboard that shows the running revenue of the current hour wants this mode: each batch sends the new value of every hour and country that received orders. With a watermark, state for completed hours is removed, but the rows written for an hour are provisional, and a later batch may write that hour again with a new value.
• Append writes each row once, when it is final and will never change. For an aggregation, that requires a watermark, which is why Spark rejects our query in append mode without one. Our table receives each hour exactly once, at the earliest ten minutes after the hour ends; nothing in it is ever revised.

Not every sink accepts every mode. The file sink, which our Parquet table uses, accepts only append, because files once written are never rewritten. Kafka accepts all three modes. So does foreachBatch, because it leaves the writing to you: instead of naming a sink, you pass foreachBatch a function you write, and Spark calls that function once per batch with two arguments, the batch's output as an ordinary DataFrame and the batch's number, its batch id. The function can then write the rows any way it likes, for example with a MERGE into a table.

The state store#

State is split into partitions the same way as a shuffle, by hashing the grouping key, here the hour and the country: the sum so far for (10:00, DE) always lives in the same state partition, so the task that receives the new orders for that key finds it there. The number of state partitions is spark.sql.shuffle.partitions at the moment the query first starts (200 by default). Spark records that number in the checkpoint and reuses it on every restart, because the existing state was partitioned that way. If you later set spark.sql.shuffle.partitions to 400 and restart the query from the same checkpoint, its state keeps 200 partitions; only a query started with a new checkpoint, and therefore with empty state, gets 400. Shuffles that don't feed the state do take the new value.

Each state partition lives in a state store on an executor: a key-value store that the task for that partition reads and updates during every batch. At the end of each batch, the store saves what it needs to the query's checkpoint directory, the checkpointLocation from our code: a directory on storage that outlives the executors, such as HDFS or an object store like S3.

Between batches, each state partition stays on its executor. A state store coordinator on the driver records which executor holds each state partition, and the scheduler prefers to send the next batch's task for that partition to the same executor, where the state is already loaded. If that executor is lost, or has no free core within a few seconds, another executor rebuilds the partition from the copy in the checkpoint.

Spark offers two implementations of the state store. They differ in where they keep the state while the query runs, in what they write to the checkpoint at the end of each batch, and therefore in how a lost partition is rebuilt.

The default store keeps each partition's state as a hash map in the executor's memory. At the end of each batch, it writes only the changes the task made, the keys it added, updated or removed, to the checkpoint directory as a delta file. A background thread on the executor, which runs every minute, writes the map from memory to the checkpoint as a full snapshot of the partition once more than 10 deltas have accumulated since the last snapshot (by default). To rebuild a partition, another executor loads the latest snapshot and then replays the deltas written after it: it applies their changes one after another, in batch order, until the map is as the last batch left it.

The alternative keeps the state in RocksDB, an embedded key-value database, on the executor's local disk, which suits state too large for memory. RocksDB never modifies a data file once it has written it: new and changed keys go into new files, and RocksDB merges older files into new ones in the background.

At the end of each batch, the store copies to the checkpoint directory the data files that appeared since its last copy. Files that haven't changed are already there from earlier batches and aren't copied again, so the state after a batch is spread over files copied by several batches. The store therefore also writes, for each batch, a manifest: a small zip file that holds RocksDB's own bookkeeping files and a list of every data file this version of the state needs, whichever batch copied it. The files a manifest names are the complete state as its batch left it. To rebuild a partition, another executor reads the latest manifest, downloads the files it names and opens them with RocksDB; there are no changes to replay.

The price is paid at the end of each batch. Before the store can copy anything, RocksDB writes the changes it still holds in memory to a new data file, so every batch adds a small file for each state partition it changed; and after RocksDB has merged older files, the merged file it has to copy can be large. The batch isn't finished until all of that is copied, so some batches take noticeably longer than others.

With changelog checkpointing (spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled, off by default), the store instead writes only a small file of the batch's changes, much like the default store's delta file. Every 10 batches by default, the end of a batch also writes RocksDB's memory to new data files on local disk, which takes little time next to a copy; copying those files and a manifest to the checkpoint is left to a background thread. The end of each batch becomes short and steady, which matters when results must appear within seconds, or when long copies would make batches fall behind their trigger. The cost moves to recovery: rebuilding a partition replays the changes made since the last full copy, up to 10 batches' worth, and more if the background copies have fallen behind.

Side by side:

The store is chosen with spark.sql.streaming.stateStore.providerClass before a query first starts; Spark records the choice in the checkpoint, and a restarted query keeps the store it started with.

Checkpointing#

The checkpoint directory, /checkpoints/revenue_hourly in our code, is where the query writes down everything it needs to continue after a stop: how far it has read, which batches it has finished, and its state. It sits on storage that outlives the driver and the executors, and it holds five entries:

/checkpoints/revenue_hourly/
  offsets/       one file per batch, written before the batch runs: where it reads up to, and its watermark
  commits/       one file per batch, written once the batch has finished
  state/         the state stores' files: one directory per stateful operator and state partition
  sources/0/0    where the query started reading Kafka, recorded at its first start
  metadata       the query's id

How far the query got: offsets/ and commits/. These two directories hold one small file per batch, named after the batch's number: offsets/74 and commits/74 belong to batch 74. Before batch 74 reads anything, the driver writes offsets/74: for each partition of orders, the offset up to which the batch reads (it starts where batch 73 stopped), and the watermark the batch works with. The batch then runs: its tasks read exactly those offsets, update the state, pass the hours that became final to the sink, which writes them, and only then save the state's changes. When all of that is done, the driver writes commits/74. At any moment, the two directories therefore say exactly how far the query got: a batch with a file in commits/ has finished, and a batch with a file only in offsets/ started but may not have finished. The offsets live here, not in Kafka: Spark doesn't rely on Kafka's own record of what a consumer has read.

The state: state/. Under state/0/, for our query's single stateful operator (number 0), there is one directory per state partition: 0 to 199 here, because our query has 200 state partitions, the default of spark.sql.shuffle.partitions when it first started; a query started with a different value has that many directories instead. Each holds the files the state store writes there, as the previous section described: delta files and snapshots for the default store, data files and manifests for RocksDB.

Where reading began: sources/0/0. The first batch of a new query has no earlier batch to continue from, so it starts where startingOffsets says. Our query sets none, which for a streaming query means latest: the end of each partition at the moment the query first starts, so orders that were already in Kafka are not read. (earliest would start with every order Kafka still keeps, and a JSON string can give an offset per partition.) "Latest" keeps moving, though: if the query failed before its first batch finished and were restarted a few minutes later, a second lookup would start later and skip the orders in between. So at the first start, the Kafka source looks up the offsets once and writes them to sources/0/0, and a restart before the first batch has finished reads them from there instead of looking them up again. After the first batch, startingOffsets plays no role: a restarted query always continues from the checkpoint, and partitions added to the topic later are read from their beginning.

The path itself: Spark gives each streaming source of a query a directory under sources/, numbered 0, 1 and so on in the order it finds them in the query's plan, so our only streaming source has sources/0. Inside it, the Kafka source writes one file, named 0 because it is the first and only one. (Since Spark 4.2, the still experimental readStream.name("orders") names a source, and its directory becomes sources/orders.)

Which query this is: metadata. It is a single file that holds the query's id, a random identifier (a UUID) that Spark creates at the query's first start and reads back at every restart. The Spark UI and the query's progress reports show it, so a restarted query is recognizable as the same query; each start also gets a new run id, which tells the starts apart. Spark also uses the id to keep two runs of one query from writing the same checkpoint: if the query is started again while it is still running in the same application, Spark stops the running copy first (spark.sql.streaming.stopActiveRunOnRestart, true by default).

Recovery#

Suppose the driver crashes while batch 74 is running, and the query is restarted with the same checkpoint location. Spark first reads the checkpoint: the last file in offsets/ is offsets/74, but the last one in commits/ is commits/73, so batch 74 started and didn't finish. Spark runs batch 74 again, with exactly the offsets and the watermark recorded in offsets/74, not with whatever has arrived in Kafka since. Each state partition is restored to the version batch 73 left behind: the default store loads its latest snapshot and replays the deltas written after it, and the RocksDB store downloads the files named in batch 73's manifest. The re-run therefore sees the same input and the same state as the first attempt, and every order is counted exactly once in the state.

If the crash had come after commits/74 was written, there would be nothing to redo: the restart would plan batch 75, starting where batch 74 stopped.

Whether the output is also written exactly once depends on the sink: it must recognize a batch it has already written, because the first attempt of batch 74 may have finished writing before the crash. Sinks differ here:

• The file sink, which writes our Parquet table, recognizes it. It keeps its own log of the files each batch produced, in a _spark_metadata directory inside the table, and skips a batch that the log already holds. Files that an interrupted attempt left behind are not in the log, so Spark ignores them when it reads the table; another engine that lists the table's directory sees them.
• Delta Lake and Iceberg tables recognize it too: each commit records the query's id and the batch id, and a batch that is already committed is skipped.
• The Kafka sink doesn't. A batch that runs again writes its records again, and readers of the output topic see them twice (at-least-once delivery).
• With foreachBatch, Spark calls the function again for the repeated batch, with the same batch id, and can't know what the first call already wrote. The function has to make a repeated batch harmless itself, for example by writing with MERGE instead of appending, or by skipping a batch id it has already written.

A lost executor is a smaller event. Its running tasks fail and are retried on other executors, which rebuild the state partitions they need from the checkpoint; the batch then completes as usual.

Each micro-batch costs from a few hundred milliseconds to seconds of planning, scheduling and logging, however few events it carries, which is the price of this reliability. The streaming model gets a post of its own. In short, it covers triggers, sources and sinks with their guarantees, output modes, watermarks and late data, the stateful operators, and what exactly-once means from Kafka to the sink. How the loop runs inside gets another: a thread on the driver for each query, the offset and commit logs, the state store providers and their coordinator, and real-time mode.

Running on a cluster#

So far we took the cluster for granted: a driver and 10 executors with 4 cores and 8 GB of heap each. Two pieces put an application there. spark-submit is the command-line tool that launches an application, and the cluster manager hands out the machines' resources. On YARN, the cluster manager of Hadoop clusters, a central ResourceManager decides where containers run, and a NodeManager on each machine starts and watches them. On Kubernetes, the cluster's control plane schedules pods onto machines. A launch on YARN looks like this:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 8g \
  revenue_job.py

--master yarn chooses the cluster manager. --num-executors, --executor-cores and --executor-memory ask for 10 executors with 4 cores and 8 GB of heap each, which, with the overhead, makes the 8.8 GB requests we computed earlier (9 GB containers on YARN). The last argument, revenue_job.py, is the Python file with our batch query: the program that the driver runs.

--deploy-mode decides where the driver runs. In client mode, the driver runs inside the spark-submit process, on the machine where you started it. That suits interactive work and notebooks, where you want to see results directly, but the application ends when that process ends, for example when you close your session. In cluster mode, the cluster manager starts the driver inside the cluster (on YARN, in the application's first container; on Kubernetes, in a driver pod), and spark-submit only reports the application's status until it finishes. Production jobs usually run in cluster mode, so that they don't depend on the machine that started them.

Either way, once the driver runs, it asks the cluster manager for executors. On YARN, each executor gets a container on one of the cluster's machines. On Kubernetes, each executor is a pod, which the driver creates itself through the Kubernetes API. The cluster manager finds room for the executors (and, in cluster mode, the driver); as we saw, it never decides which task runs where.

How many executors to run is the first sizing decision. A fixed number suits work that keeps every core busy. Our batch query nearly does: its scan runs 800 tasks, and only its final aggregation runs a single one, because AQE merged its 200 partitions into one, as we saw earlier. That last stage takes seconds, so the 9 idle executors cost little, and they still hold the shuffle files it reads. The waste grows when the work needs fewer executors for longer: an application that runs queries on and off all day, such as a notebook session, holds all its executors between queries, and a job whose later stages run for minutes on a few tasks keeps most of its executors idle.

Dynamic allocation (spark.dynamicAllocation.enabled, off by default) adapts the number of executors to the work. When tasks have waited a second for a free slot, the driver asks the cluster manager for more executors, and asks again every second, for more each time, while tasks keep waiting. When an executor has had no task for a minute, the driver releases it. An executor that holds cached data is kept instead, by default for as long as the data stays cached; one that holds shuffle files may be kept too, as the next paragraph explains. Our streaming query is a different case: it runs a batch every minute, so its executors rarely sit idle for a whole minute and are rarely released.

Releasing an executor raises a question we met in the failure example: a later stage may still need the shuffle files it wrote. Spark has three answers of its own:

• On YARN and standalone clusters, an external shuffle service (spark.shuffle.service.enabled), running outside the executors on each machine, keeps serving the files of executors that are gone, so an executor with shuffle files is released after its idle minute like any other.
• Without a shuffle service, shuffle tracking (spark.dynamicAllocation.shuffleTracking.enabled, on by default) keeps every executor that holds shuffle files the driver still tracks. The driver stops tracking a shuffle only once nothing in the program refers any more to the query that produced the shuffle, and the driver's garbage collector has freed that query's objects; Spark forces such a garbage collection every 30 minutes. An idle executor can therefore stay long after the last stage that read its files, unless spark.dynamicAllocation.shuffleTracking.timeout, which sets no limit by default, releases it sooner.
• With decommissioning turned on (spark.decommission.enabled and spark.storage.decommission.enabled, both off by default), an executor that is about to be removed first copies its shuffle files to other executors, or to a fallback location on shared storage if one is configured. This releases idle executors early only when shuffle tracking is turned off: with both on, shuffle tracking still keeps every executor whose files it tracks, and Spark warns about it when the application starts.

Apache Spark has no external shuffle service for Kubernetes, so there only the last two apply.

The post on shuffles shows how the service and the copies serve the files, and the post on deploying Spark how dynamic allocation and decommissioning behave on YARN and Kubernetes.

Settings like these can come from three places, and the one closest to your code wins: values set in your program when it creates the session, as in SparkSession.builder.config("spark.executor.memory", "8g").getOrCreate() (or on SparkConf, its older counterpart), then spark-submit flags and --conf options, then the spark-defaults.conf file of the Spark installation that spark-submit runs from.

Not every run needs a cluster. With master("local[4]"), Spark runs in local mode: the driver and a single executor share one JVM on your machine, and 4 threads serve as task slots; the planner, AQE, shuffles and the Spark UI work as on a cluster, which makes local mode the place to develop and test. A test runs the query on a few rows built inside the test, with the query moved into a function:

import pytest
from datetime import date
from pyspark.sql import SparkSession, functions as F
from pyspark.testing import assertDataFrameEqual

def revenue_by_country(orders, customers):
    return (orders.where(F.col("order_date") >= "2026-01-01")
            .join(customers.where(F.col("segment") == "B2B"), "customer_id")
            .groupBy("country")
            .agg(F.sum("amount").alias("revenue")))

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[4]").getOrCreate()

def test_revenue_by_country(spark):
    orders = spark.createDataFrame(
        [(1, date(2026, 3, 1), 100.0), (1, date(2025, 12, 31), 70.0), (2, date(2026, 3, 2), 50.0)],
        "customer_id long, order_date date, amount double")
    customers = spark.createDataFrame(
        [(1, "B2B", "DE"), (2, "B2C", "FR")],
        "customer_id long, segment string, country string")
    expected = spark.createDataFrame([("DE", 100.0)], "country string, revenue double")
    assertDataFrameEqual(revenue_by_country(orders, customers), expected)

The fixture creates one session that all tests share, so they pay its start-up once, and assertDataFrameEqual compares the rows and the schema, ignoring the order of the rows by default. What local mode can't show is what depends on a cluster: shuffle data never crosses a network, there is no executor to lose, and the driver and the tasks share one heap.

Many teams never run a cluster of their own. Managed platforms, such as Databricks, Amazon EMR, Google Cloud's Dataproc and Microsoft Fabric, run the cluster manager, scale the executors and keep the history server for you, and Databricks' serverless compute, for example, runs your code through Spark Connect, described next. Each platform ships its own build of Spark, with defaults and features of its own, so what this series describes, Apache Spark 4.2, can differ there: Microsoft Fabric, for example, can run queries on a native engine written in C++ instead of in the JVM.

A self-managed cluster also has to be secured by whoever runs it: by default, Spark's processes talk to each other without authentication (spark.authenticate) or encryption (spark.network.crypto.enabled), and anyone who can reach the Spark UI's port can see its queries and settings (spark.acls.enable).

Deploying Spark gets a post of its own. In short, it covers the cluster managers and deploy modes; YARN; Kubernetes and the Spark Operator, an add-on that manages each application as a Kubernetes object with a status and can restart it when its driver fails; dynamic allocation and decommissioning on YARN and on Kubernetes; container images; and the order in which configuration is applied.

Spark Connect#

Both deploy modes share one assumption: your program runs on the same machine as the driver. In classic PySpark, which everything so far has described, your Python program and the driver JVM run side by side on one machine and talk through Py4J: your laptop or a notebook server in client mode, the driver's container in cluster mode. That machine needs Java and the full PySpark package with Spark's JVM libraries, and the executors must be able to connect to it.

Spark Connect, added in Spark 3.4, separates your program from the driver. The driver becomes a long-running server on the cluster, the Spark Connect server: an ordinary Spark application, running on YARN or Kubernetes with executors like any other, that also accepts connections from clients, each in a session of its own. Your Python program becomes a thin client: it needs no Java and none of Spark's JVM libraries, only the pyspark-client Python package, and instead of starting a JVM it connects to the server by its address. From then on, the DataFrame API looks the same as before.

What the client sends is a plan, not data. Each transformation on the client, such as where or join, only adds to an unresolved plan: the query as written, before the analyzer has checked that its tables and columns exist. An action sends that plan to the server as a Protocol Buffers message, a compact binary encoding, over gRPC, a framework for calls between processes. On the server, the plan goes through the same analyzer, optimizer, physical planner and schedulers we followed with our query, and the executors run it. The answer comes back as a stream of messages. The rows of a result, from collect() or toPandas(), arrive as Arrow batches, which the client turns into Python rows or a pandas DataFrame. An action that returns no rows, such as our saveAsTable, gets back only progress messages, with the number of completed tasks per stage, and a final message that marks it complete.

Our query's saveAsTable sends the whole query as one message. Every name in it is still a string, and the operators are the ones we wrote, not the ones Spark will run: the join has no strategy yet, and the filter isn't in the scan. On the server, the analyzer resolves the names, and the optimizer and the planner make those choices, as they did for our classic query.

Many clients can share one server. Each gets a session of its own, with its own temporary views and SQL settings, while all of them share the server's executors. The separation covers names and settings, not security: the code of every client, its UDFs included, runs on the same executors. A client also can't change settings that affect everyone, such as the executors' memory. Python UDFs still work: the client serializes the function into the plan, and the executors run it in Python workers as described in the section on Python, which requires the same Python minor version on the client and on the cluster. What the client lacks is access to the driver's JVM: it talks to the server only in plans, so there is no SparkContext, the driver's object that every RDD belongs to (our .rdd calls rely on it too), no RDD API and no Py4J access to Java libraries. Since Spark 4.0, spark.api.mode=connect lets an ordinary application use Spark Connect by starting a Spark Connect server for that application alone; code that calls Java libraries from Python, such as Iceberg's Java API, needs the default, classic.

A Map of the Spark Series#

We have now met every topic of the series. The Spark posts are organized by mechanism, and the following table is their index: what each one covers, so that you can go straight to the post that answers a given question.

When to Use Spark, and When Not#

Spark earns its overhead when the data or the computation does not fit on one machine, when many jobs must share a cluster, or when you need its ecosystem of connectors and table formats (Iceberg, Delta Lake, Kafka, JDBC sources, cloud object stores) with one execution model for batch and streaming. It is at its best on work that is naturally parallel and long-running: joins and aggregations over hundreds of gigabytes to petabytes, backfills, and maintenance of large Iceberg tables.

It is the wrong tool more often than people admit, because every Spark application pays fixed costs that do not shrink with the data:

• a JVM driver to start;
• executor start-up, which takes minutes on Kubernetes when nodes must be provisioned;
• task scheduling, launch and result reporting, typically milliseconds to tens of milliseconds per task;
• query planning and code generation on the driver;
• a write to local disk and a fetch over the network at every shuffle, however little data it moves;
• the commit of the output: renames and a catalog update.

For data up to tens of gigabytes, and often beyond, a single-node engine such as DuckDB or Polars is usually faster, cheaper and simpler to operate. Both are vectorized: like NumPy in a pandas UDF, they run compiled loops over batches of a column's values. On one machine there is no cluster to run, no serialization between processes, and no shuffle through local disk and the network (DuckDB and single-node analytics gets a post of its own). Miles Cole's 2024 benchmark summarizes the trade-off well: at 10 GB, DuckDB and Polars were the fastest, while at 100 GB, the larger of the two sizes he tested, Spark led in most steps. His Spark ran on Microsoft Fabric with its Native Execution Engine, a vectorized engine written in C++ that open-source Spark doesn't have. The sensible position is to choose the engine per workload rather than per platform.

Several other workloads fit Spark poorly:

• Low-latency event processing. Micro-batch Structured Streaming has latencies from hundreds of milliseconds to seconds. Spark 4.1 added a real-time mode for sub-second latency, but event-at-a-time engines such as Flink or Kafka Streams remain the default for complex low-latency stateful logic (stream processing gets a series of its own).
• Point lookups and small transactional writes. Spark has no indexes of its own and no row-level locking; a database is the right tool (databases get a series of their own).
• Tiny, frequent jobs. When a job runs every hour on a few hundred new rows, most of its wall-clock time goes into starting processes, driver-side planning, code generation and commits, not into processing data.

Before replacing Spark with another engine, find out where the time actually goes. A Spark job that spends 80 percent of its time listing object storage or committing to a catalog will not become much faster on a different engine, whereas a job that shuffles a 2 TB table three times may become several times faster with a better plan in Spark itself.

Summary#

Spark is easiest to reason about as a small number of mechanisms with clear costs:

• Driver and executors. The driver plans and schedules; executors run tasks. Losing an executor usually costs some recomputation; losing the driver ends the application, which at best starts again from the beginning.
• Partitions and tasks. One task processes one partition, so partition sizes decide parallelism, memory per task and skew.
• Shuffles and stages. Operations that need all rows with the same key in one place, such as joins and aggregations, need an all-to-all exchange through local disk, unless the rows are already split by the key or one side of a join is broadcast; each exchange ends a stage, and it is usually the most expensive part of a job.
• Lineage. When an executor is lost, Spark recomputes the data lost with the executor from the steps that produced it; by default, Spark keeps no spare copies of intermediate data, though cached data can be replicated; long lineages make recovery slow, which checkpoints address.
• The SQL engine. Catalyst and AQE optimize relational plans, and Tungsten runs generated code over binary rows; this is why DataFrames outperform RDDs of objects.
• Python. PySpark code builds plans that run in the JVM; on the executors, Python runs only inside UDFs and other Python functions you pass to Spark, where every row crosses to a Python worker and back, so built-in functions come first and pandas UDFs second.
• Memory. Running tasks and cached data share one region of each executor's heap. Tasks spill to disk when memory runs short and can evict cached data that has grown beyond storage's half, so an out-of-memory error usually points to something that can't spill or to memory Spark doesn't manage.
• Streaming. A streaming query is the same plan run in micro-batches, with state kept on the executors, a watermark deciding when a result is final, and offset and commit logs that make processing exactly once inside Spark.
• Fixed costs. Planning, scheduling and process start-up are paid per query and per application, which makes Spark a poor fit for small data and tiny, frequent jobs.

The posts that follow open each of these boxes, starting with the scheduler in jobs, stages and tasks.

Further reading#

Start here#

• Cluster Mode Overview (Apache Spark): driver, executors, cluster managers and the job, stage and task vocabulary on one page.
• Introducing Apache Spark 4.0 (Fan, Rielau, van Hövell, Kwon et al., Databricks): the 4.0 release explained by its committers, from Spark Connect to ANSI mode (errors instead of silent nulls for invalid input, now on by default) and transformWithState (a new operator for custom state in streaming queries).
• Introducing Apache Spark 4.1 (Ryza, Peng, Fan et al., Databricks): Declarative Pipelines, real-time mode, and general availability for SQL scripting (variables, loops and conditions in SQL) and for VARIANT, a column type for semi-structured values such as parsed JSON.
• Introducing Apache Spark 4.2 (Fan, Neumann, Rielau et al., Databricks): metric views (views that define measures such as revenue once, to be grouped by any dimension), geospatial types, change data capture in SQL and Data Source V2, real-time mode in PySpark.
• Structured Streaming Programming Guide (Apache Spark): triggers, watermarks, output modes, the state store providers and what each sink guarantees.
• Should You Ditch Spark for DuckDB or Polars? (Miles Cole): a measured comparison of single-node engines and Spark across data sizes.

Go deeper#

• Spark 4.0.0 release notes, 4.1.0 and 4.2.0 (Apache Spark): the authoritative feature lists with JIRA links.
• Spark Core migration guide and Spark SQL migration guide (Apache Spark): the changes in behavior of each release, most with the setting that restores the old one.
• Spark Connect Overview (Apache Spark): the client/server architecture, what is supported and what is not.
• How-to: Tune Your Apache Spark Jobs (Part 1) (Sandy Ryza, Cloudera): a clear short explanation of how transformations map to stages and shuffles.
• Anatomy of Apache Spark Application (Kirill Bobrov, luminousmen): a readable tour of the driver, executors, DAGScheduler, TaskScheduler and BlockManager.
• Project Tungsten: Bringing Apache Spark Closer to Bare Metal (Reynold Xin and Josh Rosen, Databricks): why Spark moved from JVM objects to binary rows in memory it manages itself.
• Apache Spark as a Compiler: Joining a Billion Rows per Second on a Laptop (Sameer Agarwal, Davies Liu and Reynold Xin, Databricks): whole-stage code generation and the measurements behind it.
• Adaptive Query Execution: Speeding Up Spark SQL at Runtime (Wenchen Fan, Herman van Hövell and MaryAnn Xue, Databricks): how AQE coalesces shuffle partitions, switches join strategies and splits skewed partitions using runtime statistics.
• Streaming 102 (Tyler Akidau, O'Reilly): event time, windows, watermarks and triggers, explained with diagrams.
• Job Scheduling (Apache Spark): dynamic resource allocation, its timeouts, and the ways to keep shuffle files available when executors go.
• Apache Arrow in PySpark (Apache Spark): how Arrow moves data between the JVM and Python for toPandas, pandas UDFs and Arrow-optimized Python UDFs.

Under the hood#

• The Internals of Spark Core (Jacek Laskowski): class-by-class notes on the scheduler, storage, memory and shuffle; check version-specific details against the source.
• Resilient Distributed Datasets (NSDI 2012) (Zaharia et al.): lineage-based fault tolerance, the idea the whole engine rests on.
• Spark SQL: Relational Data Processing in Spark (SIGMOD 2015) (Armbrust, Xin, Lian et al.): the DataFrame API and the Catalyst design.
• Structured Streaming (SIGMOD 2018) (Armbrust, Das, Torres et al.): incremental execution of DataFrame plans and the exactly-once argument.
• MapReduce: Simplified Data Processing on Large Clusters (OSDI 2004) (Dean and Ghemawat): the model Spark generalized, and the starting point of this post's comparison with MapReduce.
• Scalability! But at what COST? (HotOS 2015) (McSherry, Isard and Murray): single-threaded programs that beat clusters, and why a distributed system should be measured against one machine.
• The Dataflow Model (VLDB 2015) (Akidau et al.): where event time, windows and watermarks, the vocabulary Structured Streaming uses, were set out.
• Efficiently Compiling Efficient Query Plans for Modern Hardware (VLDB 2011) (Thomas Neumann): compiling a pipeline of operators into one loop, the technique behind whole-stage code generation.
• SPARK-10000: Consolidate storage and execution memory management (Apache Spark JIRA): the unified memory design, in which execution and storage borrow from each other, with its design document attached.
• DAGScheduler.scala (Apache Spark source): the stage-oriented scheduler; its class comment is a good one-page description of jobs, stages and fetch-failure handling.