Execution Model: Advanced

The program below sums ride fares per city over a table of 9 rides read as 3 partitions. A groupBy needs every row for a city on the same task, so the DAGScheduler cuts the job in 2. Stage 0 runs 3 tasks that read the partitions and prepare their rows for regrouping by city; stage 1 runs 3 tasks that each gather their cities' rows and sum them. Watch the order: no stage 1 task starts until all 3 stage 0 tasks have reported success.

Count the work: 6 tasks in 2 stages, 1 job, with adaptive query execution off as in the animation. The DAGScheduler built both task sets and decided their order, and it never chose an executor. The TaskScheduler placed all 6 tasks and would have retried any that failed, and it never knew they belonged to a groupBy. When the last task of stage 1 succeeded, the DAGScheduler marked the job done and collect() returned the 3 city totals. Adaptive query execution is on in a default session since Spark 3.2, so the Jobs tab lists stage 0 as a job of its own, then as skipped in the job that returns the totals, and the 3 tiny reduce partitions may run as 1 task.

When a task throws, its executor reports the failure and the exception to the driver. The TaskScheduler keeps a manager for each task set, and that manager records the failure against the partition and puts the task back among the pending ones, so the next free core runs it as a new attempt. Each partition gets spark.task.maxFailures attempts, 4 by default. Once a single partition fails that many times, the manager gives up, the DAGScheduler aborts the stage, and the job fails.

The limit counts per partition. If 300 different tasks each fail once, the job still succeeds, provided each later attempt succeeds. Local mode allows 1 attempt unless the master is written as local[4,3], where the 3 sets the attempt limit, so a laptop test fails on the first error a cluster would retry.

The driver then logs Task 3 in stage 2.0 failed 4 times, most recent failure: Lost task 3.3 in stage 2.0. In stage 2.0, the 0 is the stage's first attempt; in task 3.3, the second 3 is the attempt number, counted from 0. Identical errors on 3 executors point at the data or the code, and a 5th attempt would fail the same way.

Stage 2's tasks read the output that stage 1's tasks left on their executors. Suppose executor 4 ran 12 of stage 1's 100 tasks and dies while stage 2 is running. The first stage 2 task that asks executor 4 for its share gets a fetch failure. Another attempt of that task would ask for the same missing output, so the attempt is not counted and the failure goes up to the DAGScheduler. It marks the 12 outputs on executor 4 as lost, resubmits stage 1 with only those 12 tasks, and when they finish, resubmits stage 2 for its unfinished partitions. The Stages tab lists the new attempts as 1 (retry 1) and 2 (retry 1). A stage gets spark.stage.maxConsecutiveAttempts tries, 4 by default, before the job aborts.

By default a retry may land on the executor where it failed, since the TaskScheduler still prefers the executor next to the data. If that machine has a bad disk, the retry fails again for a reason unrelated to your code. With spark.excludeOnFailure.enabled set to true (it is off by default), Spark stops retrying a task on an executor where it already failed, and excludes an executor or a node from the rest of the stage after repeated failures there.

Defaults move between releases, so check the configuration page for the version you run before relying on these numbers.

Task 117's original had read half its partition after 95 seconds, because its host's disk was serving reads slowly. Its copy, on a healthy executor, runs for 41 seconds and finishes at 136 seconds. The TaskScheduler marks partition 117 done and kills the original with the reason another attempt succeeded. Left alone, the original would have needed 190 seconds, so the copy saved 54.

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.

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.

from pyspark.sql import functions as F

orders = spark.table("orders")

summary = (
    orders
    .filter(F.col("region").isNotNull())
    .filter(F.col("profit").isNotNull())
    .groupBy("region")
    .agg(
        F.count("*").alias("order_count"),
        F.round(F.sum("profit"), 2).alias("total_profit"),
    )
)

summary.collect()

When a core frees up on a machine holding none of a pending task's data, the TaskScheduler can run a task there now at a worse level, or hold the task briefly in case a core frees up next to its data. Spark holds it. The wait per level is spark.locality.wait, 3s by default, with overrides in spark.locality.wait.process, spark.locality.wait.node and spark.locality.wait.rack. The timer belongs to the task set: after 3s with no launch at its current level, the set relaxes one level, and it keeps relaxing until ANY.