ShuffleMapTask — Task for ShuffleMapStage

ShuffleMapTask is a Task that computes a MapStatus, i.e. writes the result of computing records in a RDD partition to the shuffle system and returns information about the BlockManager and estimated size of the result shuffle blocks.

ShuffleMapTask is created exclusively when DAGScheduler submits missing tasks for a ShuffleMapStage.

Table 1. ShuffleMapTask’s Internal Registries and Counters
Name	Description
`preferredLocs`	Collection of TaskLocations. Corresponds directly to unique entries in locs with the only rule that when `locs` is not defined, it is empty, and no task location preferences are defined. Initialized when `ShuffleMapTask` is created. Used exclusively when `ShuffleMapTask` is requested for preferred locations.

Note	Spark uses broadcast variables to send (serialized) tasks to executors.

Creating ShuffleMapTask Instance

ShuffleMapTask takes the following when created:

stageId — the stage of the task
stageAttemptId — the stage’s attempt
taskBinary — the broadcast variable with the serialized task (as an array of bytes)
Partition
Collection of TaskLocations
localProperties — task-specific local properties
serializedTaskMetrics — the serialized FIXME (as an array of bytes)
jobId — optional ActiveJob id (default: undefined)
appId — optional application id (default: undefined)
appAttemptId — optional application attempt id (default: undefined)

ShuffleMapTask calculates preferredLocs internal attribute that is the input locs if defined. Otherwise, it is empty.

Note	`preferredLocs` and `locs` are transient so they are not sent over the wire with the task.

ShuffleMapTask initializes the internal registries and counters.

Writing Records (After Computing RDD Partition) to Shuffle System — `runTask` Method

runTask(context: TaskContext): MapStatus

Note	`runTask` is a part of Task contract to…FIXME

runTask computes a MapStatus (which is the BlockManager and an estimated size of the result shuffle block) after the records of the Partition were written to the shuffle system.

Internally, runTask uses the current closure Serializer to deserialize the taskBinary serialized task (into a pair of RDD and ShuffleDependency).

runTask measures the thread and CPU time for deserialization (using the System clock and JMX if supported) and stores it in _executorDeserializeTime and _executorDeserializeCpuTime attributes.

Note	`runTask` uses `SparkEnv` to access the current closure `Serializer`.

Note	The `taskBinary` serialized task is given when `ShuffleMapTask` is created.

runTask requests ShuffleManager for a ShuffleWriter (given the ShuffleHandle of the deserialized ShuffleDependency, partitionId and input TaskContext).

Note	`runTask` uses `SparkEnv` to access the current `ShuffleManager`.

Note	The `partitionId` partition is given when `ShuffleMapTask` is created.

runTask gets the records in the RDD partition (as an Iterator) and writes them (to the shuffle system).

Note	This is the moment in `Task`'s lifecycle (and its corresponding RDD) when a RDD partition is computed and in turn becomes a sequence of records (i.e. real data) on a executor.

runTask stops the ShuffleWriter (with success flag enabled) and returns the MapStatus.

When the record writing was not successful, runTask stops the ShuffleWriter (with success flag disabled) and the exception is re-thrown.

You may also see the following DEBUG message in the logs when the ShuffleWriter could not be stopped.

DEBUG Could not stop writer

`preferredLocations` Method

preferredLocations: Seq[TaskLocation]

Note	`preferredLocations` is a part of Task contract to…FIXME

preferredLocations simply returns preferredLocs internal property.

ShuffleMapTask — Task for ShuffleMapStage