The Driver as Bottleneck
Concepts covered: Spark Execution Model
Everything in the last 3 sections runs in 1 process. The DAGScheduler handles its events one at a time on a single thread, and the TaskScheduler shares the driver's JVM, tracking every executor and every task attempt. Executors scale out and the driver does not, so a job can add 500 cores and still be capped by how fast a single machine schedules, tracks and receives the work those cores do.
- Builds and serializes every task before sending it to an executor
- Takes a status update each time a task attempt starts, succeeds, fails or is killed
- Tracks where each earlier-stage task left its output, with a size per next-stage partition
- Feeds every scheduler event to the UI through a queue holding 10,000 events by default, which drops the overflow
- Receives each task's result for actions such as collect() and count()
Tiny tasks and the driver's serial work
Suppose a stage launches 20,000 tasks, each doing 30ms of real work, on a cluster with 200 cores. Assume each task also costs the driver about 2ms of serial work: launching it, handling its status updates and recording its output. Price both sides:
- 3 s
- 40 s
The cores could clear their share quickly, but the driver cannot hand out and account for tasks that fast, so they spend most of the stage waiting for their next task. The 2ms is an assumption for the arithmetic, and the shape holds at any value: once a task's own work approaches what the driver spends on it, more cores change nothing. Spark's tuning guide says tasks as short as 200ms run efficiently. Well below that, merge the work into fewer, larger tasks.
The record of stage output grows with task counts on both sides of a boundary. In a stage of 50,000 tasks feeding a stage of 10,000 partitions, each task reports a status covering all 10,000 partitions. Above 2,000 partitions, each status keeps an average size and exact sizes for only the unusually large blocks, instead of every size. The driver still holds 50,000 such records and serves them to every executor that runs the next stage.
Results too large for the driver
An action's results end on the driver. For count() each task sends back 1 number; for collect() each task serializes its rows and the driver holds all of them in memory at once. The setting spark.driver.maxResultSize, 1g by default, guards this: when an action's serialized results pass it, the job fails with Total size of serialized results of 412 tasks (1025.3 MiB) is bigger than spark.driver.maxResultSize (1024.0 MiB). Raising the limit moves the failure into the driver's own memory.
Below, a filter with no stage boundary runs on 4 partitions and its rows are collected. Watch the last step, where every task's matching rows travel to the driver.
from pyspark.sql import functions as F
us_events = events.filter(F.col("country") == "US")
us_events.collect()| event_id | user_id | country | event_type |
|---|---|---|---|
| 1 | 101 | US | click |
| 2 | 102 | DE | view |
| 3 | 103 | US | view |
| 4 | 104 | BR | click |
| 5 | 105 | US | purchase |
| 6 | 106 | US | view |
| 7 | 107 | DE | click |
| 8 | 108 | US | click |
| 9 | 109 | BR | view |
| 10 | 110 | US | view |
| 11 | 111 | DE | purchase |
| 12 | 112 | US | click |
Here 7 rows arrive. Over a year of events, the same program pushes every matching row through 1 network link into 1 heap while the executors that filtered them sit idle. Aggregating first keeps the work on the executors: a groupBy before the action adds a stage boundary, and the driver receives a row per group instead of a row per event.
Run the cell. It counts orders and sums profit per region on the executors, so the driver receives a few small rows. Then delete the .groupBy line and the .agg(...) block and run it again to see how many rows the driver receives instead.
Telling a driver bottleneck apart
| What you see | Where | What it points to |
|---|---|---|
| Executors idle while a stage still has pending tasks | Event Timeline on the stage page | Driver handing out tasks too slowly, or a locality wait |
| Thousands of tasks with a median duration under 200ms | Summary metrics on the stage page | Tasks too small for their scheduling cost |
| One driver core at 100% CPU | Host metrics for the driver machine | Scheduler thread saturated |
| `Dropping event from queue` warnings | Driver log | Events lost; UI numbers may be incomplete |
| maxResultSize error or driver out of memory at the action | Driver log | Too much data returned to the driver |
A job's main stage runs 60,000 tasks with a median of 40ms each. Doubling the executors from 50 to 100 left its runtime unchanged. What is most likely to speed it up?
- Give each executor more memory so each task finishes faster and the stage needs fewer waves.
- Turn on speculation so the slowest of the 60,000 tasks get copies on idle executors.
- Merge the work into a few thousand larger tasks so the driver schedules far fewer.
- Double the executors again, since tasks this short need more slots running at once.
About this lesson
This section is part of the Execution Model: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.