TaskSchedulerImpl — Default TaskScheduler

applicationId simply request SchedulerBackend for the Spark application’s id.

executorHeartbeatReceived is…​

cancelTasks cancels all tasks submitted for execution in a stage stageId.

handleSuccessfulTask simply forwards the call to the input taskSetManager (passing tid and taskResult).

handleTaskGettingResult simply forwards the call to the taskSetManager.

schedulableBuilder is a SchedulableBuilder for the TaskSchedulerImpl.

It is set up when a TaskSchedulerImpl is initialized and can be one of two available builders:

getRackForHost is a method to know about the racks per hosts and ports. By default, it assumes that racks are unknown (i.e. the method returns None).

getRackForHost is currently used in two places:

TaskSchedulerImpl takes the following when created:

TaskSchedulerImpl initializes the internal registries and counters.

TaskSchedulerImpl sets schedulingMode to the value of spark.scheduler.mode setting (defaults to FIFO).

Failure to set schedulingMode results in a SparkException:

Ultimately, TaskSchedulerImpl creates a TaskResultGetter.

initialize initializes TaskSchedulerImpl.

initialize saves the input SchedulerBackend.

initialize then sets schedulable Pool as an empty-named Pool (passing in SchedulingMode, initMinShare and initWeight as 0).

initialize sets SchedulableBuilder (based on SchedulingMode):

initialize requests SchedulableBuilder to build pools.

As part of initialization of a SparkContext, TaskSchedulerImpl is started (using start from the TaskScheduler Contract).

start starts the scheduler backend.

start also starts task-scheduler-speculation executor service.

statusUpdate finds TaskSetManager for the input tid task (in taskIdToTaskSetManager).

When state is LOST, statusUpdate…​FIXME

When state is one of the finished states, i.e. FINISHED, FAILED, KILLED or LOST, statusUpdate cleanupTaskState for the input tid.

statusUpdate requests TaskSetManager to unregister tid from running tasks.

statusUpdate requests TaskResultGetter to schedule an asynchrounous task to deserialize the task result (and notify TaskSchedulerImpl back) for tid in FINISHED state and schedule an asynchrounous task to deserialize TaskFailedReason (and notify TaskSchedulerImpl back) for tid in the other finished states (i.e. FAILED, KILLED, LOST).

If a task is in LOST state, statusUpdate notifies DAGScheduler that the executor was lost (with SlaveLost and the reason Task [tid] was lost, so marking the executor as lost as well.) and requests SchedulerBackend to revive offers.

In case the TaskSetManager for tid could not be found (in taskIdToTaskSetManager registry), you should see the following ERROR message in the logs:

Any exception is caught and reported as ERROR message in the logs:

speculationScheduler is a java.util.concurrent.ScheduledExecutorService with the name task-scheduler-speculation for speculative execution of tasks.

When TaskSchedulerImpl starts (in non-local run mode) with spark.speculation enabled, speculationScheduler is used to schedule checkSpeculatableTasks to execute periodically every spark.speculation.interval after the initial spark.speculation.interval passes.

speculationScheduler is shut down when TaskSchedulerImpl stops.

checkSpeculatableTasks requests rootPool to check for speculatable tasks (if they ran for more than 100 ms) and, if there any, requests SchedulerBackend to revive offers.

The acceptable number of task failures (maxTaskFailures) can be explicitly defined when creating TaskSchedulerImpl instance or based on spark.task.maxFailures setting that defaults to 4 failures.

removeExecutor removes the executorId executor from the following internal registries: executorIdToTaskCount, executorIdToHost, executorsByHost, and hostsByRack. If the affected hosts and racks are the last entries in executorsByHost and hostsByRack, appropriately, they are removed from the registries.

Unless reason is LossReasonPending, the executor is removed from executorIdToHost registry and TaskSetManagers get notified.

postStartHook is a custom implementation of postStartHook from the TaskScheduler Contract that waits until a scheduler backend is ready (using the internal blocking waitBackendReady).

stop() stops all the internal services, i.e. task-scheduler-speculation executor service, SchedulerBackend, TaskResultGetter, and starvationTimer timer.

defaultParallelism simply requests SchedulerBackend for the default level of parallelism.

When executed, you should see the following INFO message in the logs:

submitTasks creates a TaskSetManager (for the input taskSet and acceptable number of task failures).

submitTasks registers the TaskSetManager per stage and stage attempt id (in taskSetsByStageIdAndAttempt).

If there is more than one active TaskSetManager for the stage, submitTasks reports a IllegalStateException with the message:

submitTasks makes sure that the requested resources, i.e. CPU and memory, are assigned to the Spark application for a non-local environment.

When submitTasks is called the very first time (hasReceivedTask is false) in cluster mode only (i.e. isLocal of the TaskSchedulerImpl is false), starvationTimer is scheduled to execute after spark.starvation.timeout to ensure that the requested resources, i.e. CPUs and memory, were assigned by a cluster manager.

Every time the starvation timer thread is executed and hasLaunchedTask flag is false, the following WARN message is printed out to the logs:

Otherwise, when the hasLaunchedTask flag is true the timer thread cancels itself.

In the end, submitTasks requests the current SchedulerBackend to revive offers (available as backend).

createTaskSetManager creates a TaskSetManager (passing on the reference to TaskSchedulerImpl, the input taskSet and maxTaskFailures, and optional BlacklistTracker).

handleFailedTask notifies taskSetManager that tid task has failed and, only when taskSetManager is not in zombie state and tid is not in KILLED state, requests SchedulerBackend to revive offers.

taskSetFinished looks all TaskSets up by the stage id (in taskSetsByStageIdAndAttempt registry) and removes the stage attempt from them, possibly with removing the entire stage record from taskSetsByStageIdAndAttempt registry completely (if there are no other attempts registered).

taskSetFinished then removes manager from the parent’s schedulable pool.

You should see the following INFO message in the logs:

executorAdded just notifies DAGScheduler that an executor was added.

waitBackendReady waits until a SchedulerBackend is ready.

waitBackendReady keeps checking the status every 100 milliseconds until SchedulerBackend is ready or the SparkContext is stopped.

If the SparkContext happens to be stopped while waiting, waitBackendReady reports a IllegalStateException:

resourceOffers takes the resources offers (as WorkerOffers) and generates a collection of tasks (as TaskDescription) to launch (given the resources available).

Internally, resourceOffers first updates hostToExecutors and executorIdToHost lookup tables to record new hosts and executors (given the input offers).

For new executors (not in executorIdToRunningTaskIds) resourceOffers notifies DAGScheduler that an executor was added.

resourceOffers requests BlacklistTracker to applyBlacklistTimeout and filters out offers on blacklisted nodes and executors.

resourceOffers then randomly shuffles offers (to evenly distribute tasks across executors and avoid over-utilizing some executors) and initializes the local data structures tasks and availableCpus (as shown in the figure below).

resourceOffers takes TaskSets in scheduling order from top-level Schedulable Pool.

For every TaskSetManager (in scheduling order), you should see the following DEBUG message in the logs:

Only if a new executor was added, resourceOffers notifies every TaskSetManager about the change (to recompute locality preferences).

resourceOffers then takes every TaskSetManager (in scheduling order) and offers them each node in increasing order of locality levels (per TaskSetManager’s valid locality levels).

For every TaskSetManager and the TaskSetManager‘s valid locality level, resourceOffers tries to find tasks to schedule (on executors) as long as the TaskSetManager manages to launch a task (given the locality level).

If resourceOffers did not manage to offer resources to a TaskSetManager so it could launch any task, resourceOffers requests the TaskSetManager to abort the TaskSet if completely blacklisted.

When resourceOffers managed to launch a task, the internal hasLaunchedTask flag gets enabled (that effectively means what the name says “there were executors and I managed to launch a task”).

resourceOfferSingleTaskSet takes every WorkerOffer (from the input shuffledOffers) and (only if the number of available CPU cores (using the input availableCpus) is at least spark.task.cpus) requests TaskSetManager (as the input taskSet) to find a Task to execute (given the resource offer) (as an executor, a host, and the input maxLocality).

resourceOfferSingleTaskSet adds the task to the input tasks collection.

resourceOfferSingleTaskSet records the task id and TaskSetManager in the following registries:

resourceOfferSingleTaskSet decreases spark.task.cpus from the input availableCpus (for the WorkerOffer).

If there is a TaskNotSerializableException, you should see the following ERROR in the logs:

resourceOfferSingleTaskSet returns whether a task was launched or not.

TaskLocality represents a task locality preference and can be one of the following (from most localized to the widest):

WorkerOffer represents a resource offer with free CPU cores available on an executorId executor on a host.