executorAdded simply calls recomputeLocality method.

TaskSetManager is a Schedulable with the following implementation:

handleTaskGettingResult looks the TaskInfo for the task id tid up in taskInfos internal registry and marks it as fetching indirect task result. It then notifies DAGScheduler.

addRunningTask adds tid to runningTasksSet internal registry and requests the parent pool to increase the number of running tasks (if defined).

removeRunningTask removes tid from runningTasksSet internal registry and requests the parent pool to decrease the number of running task (if defined).

checkSpeculatableTasks checks whether there are speculatable tasks in a TaskSet.

If the TaskSetManager is zombie or has a single task in TaskSet, it assumes no speculatable tasks.

The method goes on with the assumption of no speculatable tasks by default.

It computes the minimum number of finished tasks for speculation (as spark.speculation.quantile of all the finished tasks).

You should see the DEBUG message in the logs:

It then checks whether the number is equal or greater than the number of tasks completed successfully (using tasksSuccessful).

Having done that, it computes the median duration of all the successfully completed tasks (using taskInfos internal registry) and task length threshold using the median duration multiplied by spark.speculation.multiplier that has to be equal or less than 100.

You should see the DEBUG message in the logs:

For each task (using taskInfos internal registry) that is not marked as successful yet (using successful) for which there is only one copy running (using copiesRunning) and the task takes more time than the calculated threshold, but it was not in speculatableTasks it is assumed speculatable.

You should see the following INFO message in the logs:

The task gets added to the internal speculatableTasks collection. The method responds positively.

When a TaskSetManager is a zombie, resourceOffer returns no TaskDescription (i.e. None).

For a non-zombie TaskSetManager, resourceOffer…​FIXME

It dequeues a pending task from the taskset by checking pending tasks per executor (using pendingTasksForExecutor), host (using pendingTasksForHost), with no localization preferences (using pendingTasksWithNoPrefs), rack (uses TaskSchedulerImpl.getRackForHost that seems to return "non-zero" value for YarnScheduler only)

From TaskSetManager.resourceOffer:

If a serialized task is bigger than 100 kB (it is not a configurable value), a WARN message is printed out to the logs (only once per taskset):

A task id is added to runningTasksSet set and parent pool notified (using increaseRunningTasks(1) up the chain of pools).

The following INFO message appears in the logs:

For example:

For each submitted TaskSet, a new TaskSetManager is created. The TaskSetManager completely and exclusively owns a TaskSet submitted for execution.

TaskSetManager keeps track of the tasks pending execution per executor, host, rack or with no locality preferences.

TaskSetManager computes locality levels for the TaskSet for delay scheduling. While computing you should see the following DEBUG in the logs:

Once a task has finished, TaskSetManager informs DAGScheduler.

handleSuccessfulTask records the tid task as finished, notifies the DAGScheduler that the task has ended and attempts to mark the TaskSet finished.

Internally, handleSuccessfulTask looks TaskInfo up (in taskInfos internal registry) and records it as FINISHED.

It then removes tid task from runningTasksSet internal registry.

handleSuccessfulTask notifies DAGScheduler that tid task ended successfully (with the Task object from tasks internal registry and the result as Success).

At this point, handleSuccessfulTask looks up the other running task attempts of tid task and requests SchedulerBackend to kill them. You should see the following INFO message in the logs:

If tid has not yet been recorded as successful, handleSuccessfulTask increases tasksSuccessful counter. You should see the following INFO message in the logs:

tid task is marked as successful. If the number of task that have finished successfully is exactly the number of the tasks to execute (in the TaskSet), the TaskSetManager becomes a zombie.

If tid task was already recorded as successful, you should merely see the following INFO message in the logs:

Ultimately, handleSuccessfulTask attempts to mark the TaskSet finished.

maybeFinishTaskSet notifies TaskSchedulerImpl that a TaskSet has finished when there are no other running tasks and the TaskSetManager is not in zombie state.

handleFailedTask removes tid task from the internal registry of running tasks and marks TaskInfo as finished. It decreases the number of the tid task’s copies running (in copiesRunning internal registry).

When executed, handleFailedTask first checks out the status of the tid task. If the tid task has already been marked as failed or killed (in taskInfos internal registry), handleFailedTask does nothing and quits.

If however the task has not been registered as failed or killed before, handleFailedTask unregisters the task as running and marks it as finished with state. The number of the running copies of the task (as recorded in copiesRunning internal registry) is decremented.

handleFailedTask uses the following pattern as the reason for the failure:

handleFailedTask then calculates the failure exception for the input reason, i.e. FetchFailed, ExceptionFailure, ExecutorLostFailure and other TaskFailedReasons.

handleFailedTask informs DAGScheduler that the tid task has ended (with the Task instance from tasks internal registry, the reason, and no result, i.e. null).

If the tid task has already been marked as successful (in successful internal registry) you should see the following INFO message in the logs:

If the tid task was not recorded as successful, the task is recorded as a pending task.

Unless the TaskSetManager is a zombie or the task failure should not be counted towards the maximum number of times the task is allowed to fail before the stage is aborted (i.e. TaskFailedReason.countTowardsTaskFailures is enabled), the optional TaskSetBlacklist is updated.

handleFailedTask increments numFailures for tid and makes sure that it is not equal or greater than the allowed number of task failures per TaskSet (as specified when the TaskSetManager was created).

If so, i.e. the number of task failures of tid reached the maximum value, you should see the following ERROR message in the logs:

And handleFailedTask aborts the TaskSet and then quits.

In the end, handleFailedTask attempts to mark the TaskSet as finished.

Up to spark.task.maxFailures attempts

When you start Spark program you set up spark.task.maxFailures for the number of failures that are acceptable until TaskSetManager gives up and marks a job failed.

In the following example, you are going to execute a job with two partitions and keep one failing at all times (by throwing an exception). The aim is to learn the behavior of retrying task execution in a stage in TaskSet. You will only look at a single task execution, namely 0.0.

A TaskSetManager is in zombie state when all tasks in a taskset have completed successfully (regardless of the number of task attempts), or if the taskset has been aborted.

While in zombie state, a TaskSetManager can launch no new tasks and responds with no TaskDescription to resourceOffers.

A TaskSetManager remains in the zombie state until all tasks have finished running, i.e. to continue to track and account for the running tasks.

abort informs DAGScheduler that the TaskSet has been aborted.

The TaskSetManager enters zombie state.

Finally, abort attempts to mark the TaskSet finished.

canFetchMoreResults checks whether there is enough memory to fetch the result of a task.

Internally, canFetchMoreResults increments the internal totalResultSize with the input size which is the result of a task. It also increments the internal calculatedTasks.

If the current internal totalResultSize is bigger than spark.driver.maxResultSize the following ERROR message is printed out to the logs:

The current TaskSet is aborted and canFetchMoreResults returns false.

Otherwise, canFetchMoreResults returns true.

TaskSetManager takes the following when created:

TaskSetManager initializes the internal registries and counters.

TaskSetManager requests the current epoch from MapOutputTracker and sets it on all tasks in the taskset.

You should see the following DEBUG in the logs:

TaskSetManager adds the tasks as pending execution (in reverse order from the highest partition to the lowest).

addPendingTask registers a index task in the pending-task lists that the task should be eventually scheduled to (per its preferred locations).

Internally, addPendingTask takes the preferred locations of the task (given index) and registers the task in the internal pending-task registries for every preferred location:

For a TaskLocation being HDFSCacheTaskLocation, addPendingTask requests TaskSchedulerImpl for the executors on the host (of a preferred location) and registers the task in pendingTasksForExecutor for every executor (if available).

You should see the following INFO message in the logs:

When addPendingTask could not find executors for a HDFSCacheTaskLocation preferred location, you should see the following DEBUG message in the logs:

If the task has no location preferences, addPendingTask registers it in pendingTasksWithNoPrefs.

addPendingTask always registers the task in allPendingTasks.

executorLost re-enqueues all the ShuffleMapTasks that have completed already on the lost executor (when external shuffle service is not in use) and reports all currently-running tasks on the lost executor as failed.

Internally, executorLost first checks whether the tasks are ShuffleMapTasks and whether an external shuffle service is enabled (that could serve the map shuffle outputs in case of failure).

If executorLost is indeed due to an executor lost that executed tasks for a ShuffleMapStage (that this TaskSetManager manages) and no external shuffle server is enabled, executorLost finds all the tasks that were scheduled on this lost executor and marks the ones that were already successfully completed as not executed yet.

executorLost registers every task as pending execution (per preferred locations) and informs DAGScheduler that the tasks (on the lost executor) have ended (with Resubmitted reason).

Regardless of whether this TaskSetManager manages ShuffleMapTasks or not (it could also manage ResultTasks) and whether the external shuffle service is used or not, executorLost finds all currently-running tasks on this lost executor and reports them as failed (with the task state FAILED).

executorLost recomputes locality preferences.

recomputeLocality recomputes the internal caches: myLocalityLevels, localityWaits and currentLocalityIndex.

recomputeLocality records the current TaskLocality level of this TaskSetManager (that is currentLocalityIndex in myLocalityLevels).

recomputeLocality computes locality levels (for scheduled tasks) and saves the result in myLocalityLevels internal cache.

recomputeLocality computes localityWaits (by finding locality wait for every locality level in myLocalityLevels internal cache).

In the end, recomputeLocality getLocalityIndex of the previous locality level and records it in currentLocalityIndex.

computeValidLocalityLevels computes valid locality levels for tasks that were registered in corresponding registries per locality level.

computeValidLocalityLevels walks over every internal registry and if it is not empty computes locality wait for the corresponding TaskLocality and proceeds with it only when the locality wait is not 0.

For TaskLocality with pending tasks, computeValidLocalityLevels asks TaskSchedulerImpl whether there is at least one executor alive (for PROCESS_LOCAL, NODE_LOCAL and RACK_LOCAL) and if so registers the TaskLocality.

computeValidLocalityLevels always registers ANY task locality level.

In the end, you should see the following DEBUG message in the logs:

getLocalityWait finds locality wait (in milliseconds) for a given TaskLocality.

getLocalityWait uses spark.locality.wait (default: 3s) when the TaskLocality-specific property is not defined or 0 for NO_PREF and ANY.