log4j.logger.org.apache.spark.scheduler.TaskResultGetter=DEBUGTaskResultGetter
TaskResultGetter is a helper class of TaskSchedulerImpl for asynchronous deserialization of task results of tasks that have finished successfully (possibly fetching remote blocks) or the failures for failed tasks.
|
Caution
|
FIXME Image with the dependencies |
|
Tip
|
Consult Task States in Tasks to learn about the different task states. |
|
Note
|
The only instance of TaskResultGetter is created while TaskSchedulerImpl is created.
|
TaskResultGetter requires a SparkEnv and TaskSchedulerImpl to be created and is stopped when TaskSchedulerImpl stops.
TaskResultGetter uses task-result-getter asynchronous task executor for operation.
|
Tip
|
Enable Add the following line to Refer to Logging. |
task-result-getter Asynchronous Task Executor
getTaskResultExecutor: ExecutorServicegetTaskResultExecutor creates a daemon thread pool with spark.resultGetter.threads threads and task-result-getter prefix.
|
Tip
|
Read up on java.util.concurrent.ThreadPoolExecutor that getTaskResultExecutor uses under the covers.
|
serializer Attribute
serializer: ThreadLocal[SerializerInstance]serializer is a thread-local SerializerInstance that TaskResultGetter uses to deserialize byte buffers (with TaskResults or a TaskEndReason).
When created for a new thread, serializer is initialized with a new instance of Serializer (using SparkEnv.closureSerializer).
|
Note
|
TaskResultGetter uses java.lang.ThreadLocal for the thread-local SerializerInstance variable.
|
taskResultSerializer Attribute
taskResultSerializer: ThreadLocal[SerializerInstance]taskResultSerializer is a thread-local SerializerInstance that TaskResultGetter uses to…
When created for a new thread, taskResultSerializer is initialized with a new instance of Serializer (using SparkEnv.serializer).
|
Note
|
TaskResultGetter uses java.lang.ThreadLocal for the thread-local SerializerInstance variable.
|
Deserializing Task Result and Notifying TaskSchedulerImpl — enqueueSuccessfulTask Method
enqueueSuccessfulTask(
taskSetManager: TaskSetManager,
tid: Long,
serializedData: ByteBuffer): UnitenqueueSuccessfulTask submits an asynchronous task (to task-result-getter asynchronous task executor) that first deserializes serializedData to a DirectTaskResult, then updates the internal accumulator (with the size of the DirectTaskResult) and ultimately notifies the TaskSchedulerImpl that the tid task was completed and the task result was received successfully or not.
|
Note
|
enqueueSuccessfulTask is just the asynchronous task enqueued for execution by task-result-getter asynchronous task executor at some point in the future.
|
Internally, the enqueued task first deserializes serializedData to a TaskResult (using the internal thread-local serializer).
The TaskResult could be a DirectTaskResult or a IndirectTaskResult.
For a DirectTaskResult, the task checks the available memory for the task result and, when the size overflows spark.driver.maxResultSize, it simply returns.
|
Note
|
enqueueSuccessfulTask is a mere thread so returning from a thread is to do nothing else. That is why the check for quota does abort when there is not enough memory.
|
Otherwise, when there is enough memory to hold the task result, it deserializes the DirectTaskResult (using the internal thread-local taskResultSerializer).
For a IndirectTaskResult, the task checks the available memory for the task result and, when the size could overflow the maximum result size, it removes the block and simply returns.
Otherwise, when there is enough memory to hold the task result, you should see the following DEBUG message in the logs:
DEBUG Fetching indirect task result for TID [tid]The task notifies TaskSchedulerImpl that it is about to fetch a remote block for a task result. It then gets the block from remote block managers (as serialized bytes).
When the block could not be fetched, TaskSchedulerImpl is informed (with TaskResultLost task failure reason) and the task simply returns.
|
Note
|
enqueueSuccessfulTask is a mere thread so returning from a thread is to do nothing else and so the real handling is when TaskSchedulerImpl is informed.
|
The task result (as a serialized byte buffer) is then deserialized to a DirectTaskResult (using the internal thread-local serializer) and deserialized again using the internal thread-local taskResultSerializer (just like for the DirectTaskResult case). The block is removed from BlockManagerMaster and simply returns.
|
Note
|
A IndirectTaskResult is deserialized twice to become the final deserialized task result (using serializer for a DirectTaskResult). Compare it to a DirectTaskResult task result that is deserialized once only.
|
With no exceptions thrown, enqueueSuccessfulTask informs the TaskSchedulerImpl that the tid task was completed and the task result was received.
A ClassNotFoundException leads to aborting the TaskSet (with ClassNotFound with classloader: [loader] error message) while any non-fatal exception shows the following ERROR message in the logs followed by aborting the TaskSet.
ERROR Exception while getting task result|
Note
|
enqueueSuccessfulTask is called when TaskSchedulerImpl is notified about a task that has finished successfully (i.e. in FINISHED state).
|
Deserializing TaskFailedReason and Notifying TaskSchedulerImpl — enqueueFailedTask Method
enqueueFailedTask(
taskSetManager: TaskSetManager,
tid: Long,
taskState: TaskState.TaskState,
serializedData: ByteBuffer): UnitenqueueFailedTask submits an asynchronous task (to task-result-getter asynchronous task executor) that first attempts to deserialize a TaskFailedReason from serializedData (using the internal thread-local serializer) and then notifies TaskSchedulerImpl that the task has failed.
Any ClassNotFoundException leads to the following ERROR message in the logs (without breaking the flow of enqueueFailedTask):
ERROR Could not deserialize TaskEndReason: ClassNotFound with classloader [loader]|
Note
|
enqueueFailedTask is called when TaskSchedulerImpl is notified about a task that has failed (and is in FAILED, KILLED or LOST state).
|