Task Failure and Retry
Concepts covered: Spark Execution Model
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.
Here partition 3 of stage 2 hits a malformed date near the end of its input:
| Attempt | In the error message | Executor | Result |
|---|---|---|---|
| 1st | 3.0 | exec 7 | Failed: malformed date in row 48,112 |
| 2nd | 3.1 | exec 7 | Failed: same error |
| 3rd | 3.2 | exec 12 | Failed: same error |
| 4th | 3.3 | exec 3 | Failed: same error, job aborted |
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.
4 failures, 4 different paths
| What went wrong | Handled by | Counts toward the 4? | What runs again |
|---|---|---|---|
| Your code throws on a row | TaskScheduler | Yes | That task, as a new attempt |
| Executor killed for exceeding its memory limit | TaskScheduler | Yes | Every task that was running on it |
| Executor removed for reasons outside the app, such as decommissioning | TaskScheduler | No | Every task that was running on it |
| A task cannot fetch output from the previous stage | DAGScheduler | No | The lost earlier tasks, then the stage |
A failure that crosses a stage boundary
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.
What a retry costs
A retry starts its partition from the first row; nothing a failed attempt computed survives. Take a task that normally runs 10 minutes and dies at minute 9 on a bad record near the end of its input. Each attempt holds a core for 9 minutes, so the 4th failure lands around 36 minutes after the first attempt began, assuming every retry finds a free core at once. The rest of the stage may have finished long before, leaving the cluster mostly idle while it waits for an abort.
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.
- Do
- Read the error under the last attempt before touching any retry setting
- Treat identical failures on several executors as a data or code bug
- Enable excludeOnFailure when failures cluster on 1 machine
- Test locally knowing each task gets 1 attempt there
- Avoid
- Raise spark.task.maxFailures to push a deterministic error through
- Blame your code for a stage retry before checking for a lost executor
- Ignore repeated task retries in a job that eventually succeeds
The Stages tab shows stage 5 twice, the second time as 5 (retry 1). Stage 4 also has a retry, with 9 tasks, although every task of its first attempt succeeded. What happened?
- Nine stage 4 tasks threw exceptions and were retried, so stage 5 restarted to read their new output.
- Stage 5 used up its 4 task attempts, so the DAGScheduler restarted it and rechecked part of stage 4.
- An executor died, and the TaskScheduler retried 9 stage 5 tasks on other executors as new attempts.
- An executor holding 9 of stage 4's outputs was lost; a fetch failure made the DAGScheduler rerun those 9 tasks, then stage 5.
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.