Execution Model: Beginner

A SQL database runs as 1 engine: 1 process parses the statement, plans it and scans the rows, on 1 machine. Spark splits that engine into separate processes with separate responsibilities, and a Spark application is the conversation between them. Nearly every surprising thing Spark does, from a collect() that crashes to a cluster of 100 machines running slower than a laptop, is a statement about which process was asked to do what.

collect() is the most common way rows reach the driver, alongside toPandas() and the building of a broadcast table. What makes collect() dangerous is that it pulls every row: each executor sends its whole share into the driver's memory. Spark aborts a job whose collected result passes spark.driver.maxResultSize, 1 GB by default, and once that limit is raised, a driver that receives more than it can hold dies of an out-of-memory error that reads as if the cluster failed. The cluster was fine. The wrong process was handed the rows.

Because separate processes share the job, the first question about a slow Spark job is which one is busy. A driver at 100% CPU with idle executors means the program asked the driver to do data work, usually a collect() or a large broadcast being serialised. Executors at 100% with a long queue of waiting tasks mean the job needs more capacity, or was given too little memory and is spilling to disk. Executors that never arrived mean the cluster manager had nothing to grant, and the job runs on the 2 executors it received instead of the 20 it asked for. Naming the process is most of the diagnosis, and the Spark UI's Executors tab shows all 3 states in 1 glance.

A table never arrives at an executor whole. When Spark reads a dataset it cuts the rows into partitions, each a slice that 1 task reads and processes, and the default target for a partition read from files is 128 MB, set by spark.sql.files.maxPartitionBytes. One task processes exactly 1 partition, never half of one and never 2, so the partition count of a dataset is also the task count of the work that reads it.

The animation below runs the smallest version of this. The order_items table holds 8 rows and is read as 4 partitions, so each partition carries 2 rows. The program adds a line_total column, quantity times unit price, and collects the result.

The count on read comes from the files. A folder of 80 Parquet files at 128 MB each reads as 80 partitions, and a 1 GB Parquet file reads as 8, because the reader splits a file at 128 MB boundaries. A single 10 GB gzip file reads as 1 partition, because gzip cannot be split mid-stream, and that 1 task runs on 1 core for the several minutes it takes to decompress 10 GB, while every other core in the cluster waits. Snappy-compressed Parquet splits cleanly, which is one reason it is Spark's default output format. The count can be changed after the read with repartition and coalesce, which a later lesson covers, and a grouping changes it again, as the last section shows.

The cell below runs a narrow chain on the practice seed's full order_items table: it computes each line's total, keeps the lines with a quantity of 2 or more, and selects the columns worth reading. Every step runs inside the task that holds the row, as the animation showed, with no task waiting on another.

(order_items
   .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
   .filter(F.col("quantity") >= 2)
   .select("order_id", "product_id", "quantity", "unit_price", "line_total"))

df.filter(...).groupBy(...).agg(...) runs nothing. Spark reads no rows, starts no tasks and contacts no executor, because each of those calls is a transformation: it adds a step to a plan the driver is holding and returns at once. Twenty of them can be chained and the cluster stays idle throughout. Work starts when an action is called, such as count(), collect() or write, and the action runs the whole accumulated plan in 1 burst.

The gap exists so the driver sees the whole chain before committing to a plan. Seeing all of it lets the Catalyst optimizer collapse adjacent steps and push filters down into the file read. A program that reads a 2 TB transactions table, filters to 1 day and sums the amounts never loads 2 TB: the filter travels down into the scan, the Parquet reader skips every file whose date range excludes that day, and the executors read the 6 GB that match. Had filter run eagerly on the line where it was written, 2 TB would already be in memory before the planner saw the next line.

Because nothing runs until the action, the line that throws is almost never the line that is wrong. A UDF that fails on 1 malformed row, written 30 lines up, sits in the plan untouched by any data until the write at the bottom forces execution, and the stack trace points at the write. The same holds for slowness: an action that takes 40 minutes is paying for every transformation above it, and the expensive step, usually a join or a groupBy, is somewhere in the chain.

take(20) and count() are both actions, and they cost wildly different amounts. On a chain of narrow transformations, take(20) needs 20 rows, so Spark starts 1 task on 1 partition. If the first partition has not yielded 20 rows, Spark launches another job over more partitions, which is why 1 take can show as several jobs in the UI. count() needs every row, so it runs 1 task per partition. A chain that ends in take(20) and finishes in 2 seconds says nothing about the same chain ending in count(). A notebook that feels fast because every cell ends in show() can hide a job that takes an hour in production. Spark does as much work as each action demands and no more.

The cell below filters products to in-stock rows with a known category and counts them per category. Each line adds to the plan. The editor's display is the action, and when it fires, the 2 filters, the grouping and the sort run together.

(products
   .filter(F.col("in_stock") == 1)
   .filter(F.col("category").isNotNull())
   .groupBy("category")
   .agg(F.count(F.lit(1)).alias("in_stock_count"))
   .orderBy(F.col("in_stock_count").desc()))

Change the sort direction and run again. The plan is rebuilt from the top and executed again, because Spark kept no intermediate result. The Jobs tab shows more than 1 job for this cell, since orderBy first runs a small sampling job to pick the boundaries between sorted partitions. Every action re-runs the full chain unless the program has cached a step, which is a lesson of its own.

Two settings fix the number. spark.executor.cores sets the cores per executor, and on YARN and Kubernetes it defaults to 1, so an application that never sets it runs 1 task per executor however large the machine. spark.executor.instances sets how many executors the driver asks for. Slots are the product of the 2, and the cluster manager can grant fewer executors than requested, which shrinks the product without any change to the code.

spark.executor.memory is per executor, and every core in that executor draws from the same heap. Spark sets aside about 300 MB and gives roughly 60% of the rest, set by spark.memory.fraction, to a pool that running tasks share for processing and caching, so a 16 GB heap leaves about 9.4 GB in that pool. With 4 cores, each running task works in roughly 2 GB of it, and the same 16 GB with 8 cores halves that to about 1 GB. A 128 MB Parquet partition can expand to several hundred megabytes of decoded rows in memory, and a task that groups or sorts them needs working space on top. Adding cores to an executor without adding memory is how a job that ran clean at 4 cores starts spilling to disk at 8. The slot count sets how many tasks run at once, and the memory per slot sets whether each of them finishes without spilling.

The animation runs it on 6 orders rows read as 2 partitions of 3 rows, across the regions EU, US and APAC, with the shuffle set to 2 output partitions. The first partition holds orders 1 to 3 and the second orders 4 to 6, so the 2 US rows sit in different partitions.

Before the shuffle, the map tasks turned 6 rows into 4 partial rows. Across the network travelled those 4 rows, and 3 reached the driver, so the cluster moved 7 rows for a 6-row input. Scale the table to 6 billion rows over the same 3 regions and the arithmetic holds: each map task still emits at most 3 rows, so a 3,200-partition read sends at most 9,600 partial rows across the network, a few hundred kilobytes, and 3 rows reach the driver. The same job grouped by order_id instead has nothing to combine on the map side, and all 6 billion rows cross the network. Which column a job groups by decides how much moves, and the code for the 2 jobs differs by 1 word.

Each action starts at least 1 job on the Jobs tab, and this program's collect() is job 0. Opening it shows the task count, 4 here: 2 map tasks and 2 reduce tasks. The task table on each stage's detail page shows which executor ran every task and how long it took. The Executors tab shows how many executors the application holds, with their totals. A first diagnosis is 3 reads of those pages. The job list says which actions fired, the task count says how many partitions the data was in, and the executor count says what the cluster manager granted.

The widget below asks for the same program on the practice seed's orders table, which carries nulls in region that the filter removes before the grouping. The 2 blanks are the aggregation that produces a total per region and the action that brings every result row back to the driver.

Swap collect for first and the job still runs the full grouping, because no region's total exists until every row for that region has been added up; the action only changes how many result rows travel to the driver afterwards.