作業執行源碼分析
當我們的代碼執行到了action（行動）操作之后就會觸發作業運行。在Spark調度中最重要的是DAGScheduler和TaskScheduler兩個調度器，其中，DAGScheduler負責任務的邏輯調度，
將作業拆分為不同階段的具有依賴關系的任務集。TaskScheduler則負責具體任務的調度執行。
提交作業
WordCount.scala執行到wordSort.collect()才會觸發作業執行，在RDD的源碼的collect方法觸發了SparkContext的runJob方法來提交作業。SparkContext的runJob方法經過幾次調用后，
進入DAGScheduler的runJob方法，其中SparkContext中調用DAGScheduler類runJob方法代碼如下：
def runJob[T, U: ClassTag](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],
resultHandler: (Int, U) => Unit): Unit = {
if (stopped.get()) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
...
// 調用DAGScheduler的runJob進行處理
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
// 做checkpoint，以后再說
rdd.doCheckpoint()
}
在DAGScheduler類內部會進行一列的方法調用，首先是在runJob方法里，調用submitJob方法來繼續提交作業，這里會發生阻塞，知道返回作業完成或失敗的結果；然后在submitJob方法里，創建了一個JobWaiter對象，
並借助內部消息處理進行把這個對象發送給DAGScheduler的內嵌類DAGSchedulerEventProcessLoop進行處理；最后在DAGSchedulerEventProcessLoop消息接收方法OnReceive中，接收到JobSubmitted樣例類完成模式匹配后，
繼續調用DAGScheduler的handleJobSubmitted方法來提交作業，在該方法中進行划分階段。
def submitJob[T, U](rdd: RDD[T],func: (TaskContext, Iterator[T]) => U,partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
// Check to make sure we are not launching a task on a partition that does not exist.
val maxPartitions = rdd.partitions.length
partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
throw new IllegalArgumentException(
"Attempting to access a non-existent partition: " + p + ". " +
"Total number of partitions: " + maxPartitions)
}
val jobId = nextJobId.getAndIncrement()
if (partitions.size == 0) {
// Return immediately if the job is running 0 tasks
return new JobWaiter[U](this, jobId, 0, resultHandler)
}
val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))
waiter
}
划分調度階段
Spark調度階段的划分是由DAGScheduler實現的，DAGScheduler會從最后一個RDD出發使用廣度優先遍歷整個依賴樹，從而划分調度階段，
調度階段的划分是以是否為寬依賴進行的，即當某個RDD的操作是Shuffle時，以該Shuffle操作為界限划分成前后兩個調度階段
代碼實現是在DAGScheduler的handleJobSubmitted方法中。
private[scheduler] def handleJobSubmitted(jobId: Int,finalRDD: RDD[_],func: (TaskContext, Iterator[_]) => _,partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
properties: Properties) {
// 第一步：使用最后一個RDD創建一個finalStage
var finalStage: ResultStage = null
try {
// 創建一個stage，並將它加入DAGScheduler內部的內存緩沖中, newResultStage的時候就已經得到了他所有的ParentStage
finalStage = newResultStage(finalRDD, func, partitions, jobId, callSite)
} catch {...}
// 第二步：用finalStage，創建一個Job
val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
clearCacheLocs()
logInfo("Got job %s (%s) with %d output partitions".format(
job.jobId, callSite.shortForm, partitions.length))
logInfo("Final stage: " + finalStage + " (" + finalStage.name + ")")
logInfo("Parents of final stage: " + finalStage.parents)
logInfo("Missing parents: " + getMissingParentStages(finalStage))
// 第三步：將job加入內存緩存中
val jobSubmissionTime = clock.getTimeMillis()
jobIdToActiveJob(jobId) = job
activeJobs += job
finalStage.setActiveJob(job)
val stageIds = jobIdToStageIds(jobId).toArray
val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
listenerBus.post(
SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
// 第四步：使用submitStage方法來提交最后一個stage，
// 最后的結果就是，第一個stage提交，其它stage都在等待隊列中
submitStage(finalStage)
// 提交等待的stage
submitWaitingStages()
}
// newResultStage會調用getParentStagesAndId得到所有的父類stage以及它們的id
private def newResultStage(rdd: RDD[_],func: (TaskContext, Iterator[_]) => _,partitions: Array[Int],
jobId: Int,
callSite: CallSite): ResultStage = {
// 得到所有的父stage
val (parentStages: List[Stage], id: Int) = getParentStagesAndId(rdd, jobId)
val stage = new ResultStage(id, rdd, func, partitions, parentStages, jobId, callSite)
stageIdToStage(id) = stage
updateJobIdStageIdMaps(jobId, stage)
stage
}
// 繼續調用getParentStages
private def getParentStagesAndId(rdd: RDD[_], firstJobId: Int): (List[Stage], Int) = {
val parentStages = getParentStages(rdd, firstJobId)
val id = nextStageId.getAndIncrement()
(parentStages, id)
}
private def getParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
val parents = new HashSet[Stage]
val visited = new HashSet[RDD[_]]
// We are manually maintaining a stack here to prevent StackOverflowError
// caused by recursively visiting
val waitingForVisit = new Stack[RDD[_]]
def visit(r: RDD[_]) {
if (!visited(r)) {
visited += r
for (dep <- r.dependencies) {
dep match {
// 所依賴RDD操作類型是ShuffleDependency，需要划分ShuffleMap調度階段，
// 以getShuffleMapStage方法為入口，向前遍歷划分調度階段
case shufDep: ShuffleDependency[_, _, _] =>
parents += getShuffleMapStage(shufDep, firstJobId)
case _ =>
waitingForVisit.push(dep.rdd)
}
}
}
}
// 從最后一個RDD向前遍歷這個依賴樹，如果該RDD依賴樹存在ShuffleDependency的RDD，
// 則存在父調度階段，反之，不存在
waitingForVisit.push(rdd)
while (waitingForVisit.nonEmpty) {
visit(waitingForVisit.pop())
}
parents.toList
}
//
private def getShuffleMapStage(shuffleDep: ShuffleDependency[_, _, _],firstJobId: Int): ShuffleMapStage = {
shuffleToMapStage.get(shuffleDep.shuffleId) match {
case Some(stage) => stage
case None =>
// We are going to register ancestor shuffle dependencies
getAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
if (!shuffleToMapStage.contains(dep.shuffleId)) {
// 如果shuffleToMapStage中沒有，那么就new一個shuffleMapStage
shuffleToMapStage(dep.shuffleId) = newOrUsedShuffleStage(dep, firstJobId)
}
}
// Then register current shuffleDep
val stage = newOrUsedShuffleStage(shuffleDep, firstJobId)
shuffleToMapStage(shuffleDep.shuffleId) = stage
stage
}
}
// 可以對比這個算法和getParentStages，該方法返回所有的寬依賴
private def getAncestorShuffleDependencies(rdd: RDD[_]): Stack[ShuffleDependency[_, _, _]] = {
val parents = new Stack[ShuffleDependency[_, _, _]]
val visited = new HashSet[RDD[_]]
val waitingForVisit = new Stack[RDD[_]]
def visit(r: RDD[_]) {
if (!visited(r)) {
visited += r
for (dep <- r.dependencies) {
dep match {
// 所依賴RDD操作類型是ShuffleDependency，作為划分ShuffleMap調度階段界限
case shufDep: ShuffleDependency[_, _, _] =>
if (!shuffleToMapStage.contains(shufDep.shuffleId)) {
parents.push(shufDep)
}
case _ =>
}
waitingForVisit.push(dep.rdd)
}
}
}
waitingForVisit.push(rdd)
while (waitingForVisit.nonEmpty) {
visit(waitingForVisit.pop())
}
parents
}
只看代碼還是會頭大，我們結合一個圖來講解上面的代碼：如下圖，有7個RDD，分別是rddA~rddG，它們之間有5個操作，其划分調度階段如下：
- 在SparkContext中提交運行時，會調用DAGScheduler的handleJobSubmitted進行處理，在該方法中會先找到最后一個RDD(即rddG)，並調用getParentStages方法
- 在getParentStages方法判斷rddG的依賴RDD樹中是否存在shuffle操作，在該例子中發現join操作為shuffle操作，則獲取該操作的RDD為rddB和rddF
- 使用getAncestorShuffleDependencies方法從rddB向前遍歷，發現該依賴分支上沒有其他的寬依賴，調用newOrUsedShuffleStage方法生成調度階段ShuffleMapStage0
- 使用getAncestorShuffleDependencies方法從rddB向前遍歷，發現groupByKey寬依賴操作，以此為分界划分rddC和rddD為ShuffleMapStage1, rddE和rddF為ShuffleMapStage2
- 最后生成rddG的ResultStage3。
總結，語句
finalStage = newResultStage(finalRDD, func, partitions, jobId, callSite)在生成finalStage的同時，建立起所有調度階段的依賴關系，最后通過finalStage生成一個作業實例，在該作業實例中按照順序提交調度階段進行執行。
提交調度階段
通過handleJobSubmitted方法中的submitStage(finalStage)來提交作業。在submitStage方法中調用getMissingParentStages方法獲取finalStage父調度階段，如果不存在父調度階段，則使用submitMissingTasks方法提交執行，如果存在父調度階段，
則把該調度階段存放到waitingStages列表中，同時遞歸調用submitStage。
/**
* 提交stage的方法
* stage划分算法的入口，stage划分算法是由submitStage()與getMissingParentStages()共同組成
*/
private def submitStage(stage: Stage) {
val jobId = activeJobForStage(stage)
if (jobId.isDefined) {
logDebug("submitStage(" + stage + ")")
if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
// getMissingParentStages獲取當前stage的父stage
val missing = getMissingParentStages(stage).sortBy(_.id)
logDebug("missing: " + missing)
// 直到最初的stage，它沒有父stage，那么此時，就會去首先提交第一個stage，stage0
// 其余的stage，都在waitingStages里
if (missing.isEmpty) {
logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
submitMissingTasks(stage, jobId.get)
} else {
for (parent <- missing) {
// 遞歸調用submitStage方法去提交父stage
submitStage(parent)
}
// 並且當前stage放入waitingStages等待執行的stage隊列中
waitingStages += stage
}
}
} else {
abortStage(stage, "No active job for stage " + stage.id, None)
}
}
private def getMissingParentStages(stage: Stage): List[Stage] = {
val missing = new HashSet[Stage]
val visited = new HashSet[RDD[_]]
// We are manually maintaining a stack here to prevent StackOverflowError
// caused by recursively visiting
val waitingForVisit = new Stack[RDD[_]]
def visit(rdd: RDD[_]) {
if (!visited(rdd)) {
visited += rdd
val rddHasUncachedPartitions = getCacheLocs(rdd).contains(Nil)
if (rddHasUncachedPartitions) {
// rdd的依賴
for (dep <- rdd.dependencies) {
dep match {
// 如果是寬依賴，那么就創建一個shuffleMapStage
case shufDep: ShuffleDependency[_, _, _] =>
val mapStage = getShuffleMapStage(shufDep, stage.firstJobId)
// 判斷是否可用，也就是判斷父stage有沒有結果, 看源碼可以發現就是判斷_numAvailableOutputs == numPartitions
// _numAvailableOutputs就是每個task成功后會+1
if (!mapStage.isAvailable) {
missing += mapStage
}
// 如果是窄依賴，那么將依賴的rdd放入棧中
case narrowDep: NarrowDependency[_] =>
waitingForVisit.push(narrowDep.rdd)
}
}
}
}
}
// 首先往棧中推入了stage最后一個rdd
waitingForVisit.push(stage.rdd)
while (waitingForVisit.nonEmpty) {
visit(waitingForVisit.pop())
}
missing.toList
}
同樣的，我們畫圖來講解提交調度階段，如下圖
提交任務
在submitStage中會執行submitMissingTasks方法中，會根據調度階段partition個數生成對應個數task，這些任務組成一個任務集提交到TaskScheduler進行處理。
對於ResultStage生成ResultTask，對於ShuffleMapStage生成ShuffleMapTask。
private def submitMissingTasks(stage: Stage, jobId: Int) {
...
// 為stage創建指定數量的task
// 這里有一個很關鍵，就是task最佳位置的計算
val tasks: Seq[Task[_]] = try {
stage match {
case stage: ShuffleMapStage =>
partitionsToCompute.map { id =>
// 給每個partition創建一個task
// 給每個task計算最佳位置
val locs = taskIdToLocations(id)
val part = stage.rdd.partitions(id)
new ShuffleMapTask(stage.id, stage.latestInfo.attemptId,
taskBinary, part, locs, stage.latestInfo.taskMetrics, properties)
}
case stage: ResultStage =>
val job = stage.activeJob.get
partitionsToCompute.map { id =>
val p: Int = stage.partitions(id)
val part = stage.rdd.partitions(p)
val locs = taskIdToLocations(id)
new ResultTask(stage.id, stage.latestInfo.attemptId,
taskBinary, part, locs, id, properties, stage.latestInfo.taskMetrics)
}
}
} catch {...}
if (tasks.size > 0) {
logInfo("Submitting " + tasks.size + " missing tasks from " + stage + " (" + stage.rdd + ")")
stage.pendingPartitions ++= tasks.map(_.partitionId)
logDebug("New pending partitions: " + stage.pendingPartitions)
// 提交taskSet,Standalone模式，使用的是TaskSchedulerImpl
taskScheduler.submitTasks(new TaskSet(
tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))
stage.latestInfo.submissionTime = Some(clock.getTimeMillis())
} else {
// Because we posted SparkListenerStageSubmitted earlier, we should mark
// the stage as completed here in case there are no tasks to run
// 如果調度階段不存在任務標記，則標記調度階段已經完成
markStageAsFinished(stage, None)
...
}
}
當TaskSchedulerImpl收到發送過來的任務集時，在submitTasks方法中構建一個TaskSetManager的實例，用於管理這個任務集的生命周期，而該TaskSetManager會放入系統的調度池中，根據系統設置的調度算法進行調度，
TaskSchedulerImpl.submitTasks方法代碼如下：
override def submitTasks(taskSet: TaskSet) {
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
// 為每一個taskSet創建一個taskSetManager
// taskSetManager在后面負責，TaskSet的任務執行狀況的監視和管理
val manager = createTaskSetManager(taskSet, maxTaskFailures)
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
// 把manager加入內存緩存中
stageTaskSets(taskSet.stageAttemptId) = manager
val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
ts.taskSet != taskSet && !ts.isZombie
}
if (conflictingTaskSet) {
throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
}
// 將該任務集的管理器加入到系統調度池中，由系統統一調配，該調度器屬於應用級別
// 支持FIFO和FAIR兩種，默認FIFO
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
...
}
// 在創建SparkContext，創建TaskScheduler的時候，創建了StandaloneSchedulerBackend，這個backend是負責
// 創建AppClient，向Master注冊Application的, 詳見Spark運行時消息通信
backend.reviveOffers()
}
StandaloneSchedulerBackend的reviveOffers方法是繼承於父類CoarseGrainedSchedulerBackend，該方法會向DriverEndpoint發送消息，調用makeOffers方法。在該方法中先會獲取集群中可用的Executor，
然后發送到TaskSchedulerImpl中進行對任務集的任務分配運行資源，最后提交到launchTasks方法中。CoarseGrainedSchedulerBackend.DriverEndpoint.makeOffers代碼如下：
private def makeOffers() {
// 獲取集群中可用的Executor列表
val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
// workOffers是每個Executor可用的cpu資源數量
val workOffers = activeExecutors.map { case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
}.toSeq
// 第一步：調用TaskSchedulerImpl的resourceOffers()方法，執行任務分配算法，將各個task分配到executor上去
// 第二步：分配好task到executor之后，執行自己的launchTasks方法，將分配的task發送LaunchTask消息到對應executor上去，由executor啟動並執行
launchTasks(scheduler.resourceOffers(workOffers))
}
第一步：在TaskSchedulerImpl的resourceOffers()方法里進行非常重要的步驟－－資源分配, 在分配過程中會根據調度策略對TaskSetMannager進行排序，
然后依次對這些TaskSetManager按照就近原則分配資源，按照順序為PROCESS_LOCAL NODE_LOCAL NO_PREF RACK_LOCAL ANY
def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
// Mark each slave as alive and remember its hostname
// Also track if new executor is added
var newExecAvail = false
for (o <- offers) {
executorIdToHost(o.executorId) = o.host
executorIdToTaskCount.getOrElseUpdate(o.executorId, 0)
if (!executorsByHost.contains(o.host)) {
executorsByHost(o.host) = new HashSet[String]()
executorAdded(o.executorId, o.host)
newExecAvail = true
}
for (rack <- getRackForHost(o.host)) {
hostsByRack.getOrElseUpdate(rack, new HashSet[String]()) += o.host
}
}
// 首先，將可用的executor進行shuffle
val shuffledOffers = Random.shuffle(offers)
// tasks，是一個序列，其中的每個元素又是一個ArrayBuffer
// 並且每個子ArrayBuffer的數量是固定的，也就是這個executor可用的cpu數量
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
val availableCpus = shuffledOffers.map(o => o.cores).toArray
// 從rootPool中取出排序了的TaskSetManager
// 在創建完TaskScheduler StandaloneSchedulerBackend之后，會執行initialize()方法，其實會創建一個調度池
// 這里就是所有提交的TaskSetManager，首先會放入這個調度池中，然后再執行task分配算法的時候，會從這個調度池中，取出排好隊的TaskSetManager
val sortedTaskSets = rootPool.getSortedTaskSetQueue
// 如果有新加入的Executor，需要重新計算數據本地性
for (taskSet <- sortedTaskSets) {
logDebug("parentName: %s, name: %s, runningTasks: %s".format(
taskSet.parent.name, taskSet.name, taskSet.runningTasks))
if (newExecAvail) {
taskSet.executorAdded()
}
}
// Take each TaskSet in our scheduling order, and then offer it each node in increasing order
// of locality levels so that it gets a chance to launch local tasks on all of them.
// NOTE: the preferredLocality order: PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY
var launchedTask = false
for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
do {
// 對當前taskset嘗試使用最小本地化級別，將taskset的task，在executor上進行啟動
// 如果啟動不了，就跳出這個do while，進入下一級本地化級別，一次類推
launchedTask = resourceOfferSingleTaskSet(
taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
} while (launchedTask)
}
if (tasks.size > 0) {
hasLaunchedTask = true
}
return tasks
}
第二步：分配好資源的任務提交到CoarseGrainedSchedulerBackend的launchTasks方法中，在該方法中會把任務一個個發送到Worker節點上的CoarseGrainedExecutorBacken，
然后通過其內部的Executor來執行任務
// 根據分配好的tasks，在executor上啟動相應的task
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
// 序列化
val serializedTask = ser.serialize(task)
if (serializedTask.limit >= maxRpcMessageSize) {
scheduler.taskIdToTaskSetManager.get(task.taskId).foreach { taskSetMgr =>
try {
var msg = ...
taskSetMgr.abort(msg)
} catch {
case e: Exception => logError("Exception in error callback", e)
}
}
} else {
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
// 向Worker節點的CoarseGrainedExecutorBackend發送消息執行Task
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}
}
}
我們繼續通過圖解來解釋以上代碼的調用過程，如下圖所示：
- 在提交stage中，第一次調用的是ShuffleMapStage0和ShuffleMapStage1，假設都只有兩個partition，ShuffleMapStage0是TaskSet0，
ShuffleMapStage1是TaskSet1，每個TaskSet都有兩個任務在執行。 - TaskScheduler收到發送過來的任務集TaskSet0和TaskSet1，在submitTasks方法中分別構建TaskSetManager0和TaskSetManager1，並把它們兩放到系統的調度池，
根據系統設置的調度算法進行調度（FIFO或者FAIR） - 在TaskSchedulerImpl的resourceOffers方法中按照就近原則進行資源分配，使用CoarseGrainedSchedulerBackend的launchTasks方法把任務發送到Worker節點上的
CoarseGrainedExecutorBackend調用其Executor來執行任務 - 當ShuffleMapStage2同理，ResultStage3生成的是ResultTask
執行任務
當CoarseGrainedExecutorBackend接收到LaunchTask消息時，會調用Executor的launchTask方法進行處理。在Executor的launchTask方法中，初始化一個TaskRunner來封裝任務，
它用於管理任務運行時的細節，再把TaskRunner對象放入到ThreadPool中去執行。TaskRunner.run的前半部分代碼如下：
override def run(): Unit = {
val taskMemoryManager = new TaskMemoryManager(env.memoryManager, taskId)
val deserializeStartTime = System.currentTimeMillis()
// 設置當前類加載器，使用類加載器的原因，用反射的方式來動態加載一個類，然后創建這個類的對象
Thread.currentThread.setContextClassLoader(replClassLoader)
val ser = env.closureSerializer.newInstance()
logInfo(s"Running $taskName (TID $taskId)")
// 向DriverEndpoint發送狀態更新信息
execBackend.statusUpdate(taskId, TaskState.RUNNING, EMPTY_BYTE_BUFFER)
var taskStart: Long = 0
startGCTime = computeTotalGcTime()
try {
// 反序列化
val (taskFiles, taskJars, taskProps, taskBytes) =
Task.deserializeWithDependencies(serializedTask)
// Must be set before updateDependencies() is called, in case fetching dependencies
// requires access to properties contained within (e.g. for access control).
Executor.taskDeserializationProps.set(taskProps)
updateDependencies(taskFiles, taskJars)
task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader)
task.localProperties = taskProps
task.setTaskMemoryManager(taskMemoryManager)
// If this task has been killed before we deserialized it, let's quit now. Otherwise,
// continue executing the task.
if (killed) {
throw new TaskKilledException
}
logDebug("Task " + taskId + "'s epoch is " + task.epoch)
env.mapOutputTracker.updateEpoch(task.epoch)
// Run the actual task and measure its runtime.
taskStart = System.currentTimeMillis()
var threwException = true
// value對於shuffleMapTask來說，其實就是MapStatus，封裝了shuffleMapTask計算的數據，輸出的位置
val value = try {
val res = task.run( // 具體實現在ShuffleMapTask和ResultTask中
taskAttemptId = taskId,
attemptNumber = attemptNumber,
metricsSystem = env.metricsSystem)
threwException = false
res
} finally {...}
...
對於ShuffleMapTask, 它的計算結果會寫到BlockManager之中，最終返回給DAGScheduler的是一個MapStatus。該對象管理了ShuffleMapTask的運算結果存儲到BlockManager里的相關存儲信息，而不是計算結果本身，
這些存儲信息將會成為下一階段的任務需要獲得的輸入數據時的依據。ShuffleMapTask.runTask代碼如下：
override def runTask(context: TaskContext): MapStatus = {
// 反序列化獲取rdd和rdd的依賴
// 通過broadcast variable拿到rdd的數據
val deserializeStartTime = System.currentTimeMillis()
val ser = SparkEnv.get.closureSerializer.newInstance()
val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
var writer: ShuffleWriter[Any, Any] = null
try {
// 從shuffleManager中獲取shuffleWriter
val manager = SparkEnv.get.shuffleManager
writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
// rdd.iterator傳入當前task要處理的哪個partition，執行我們自己定義的算子或者是函數
// 如果rdd已經cache或者checkpoint，那么直接讀取，否則計算，計算結果保存在本地系統的blockmanager中
writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
// 返回結果MapStatus，其實就是blockmanager相關信息
writer.stop(success = true).get
} catch {...}
ResultTask的runTask方法如下，它返回的是func函數的計算結果
override def runTask(context: TaskContext): U = {
// Deserialize the RDD and the func using the broadcast variables.
val deserializeStartTime = System.currentTimeMillis()
val ser = SparkEnv.get.closureSerializer.newInstance()
val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
func(context, rdd.iterator(partition, context))
}
獲取執行結果
對於Executor的計算結果，會根據結果的大小有不同的策略
- 生成結果大於1GB，直接丟棄，可以通過spark.driver.maxResultSize來配置
- 生成結果大小在[128MB-200kB, 1GB], 會把結果以taskId為編號存入到BlockManager中，然后把編號通過Netty發送給DriverEndpoint，Netty傳輸框架最大值和預留空間的差值
- 生成結果大小在[0, 128MB－200KB]，通過Netty直接發送到DriverEndpoint。
具體執行在TaskRunner的run方法后半部分：
// 對生成的結果序列化，並將結果放入DirectTaskResult中
val resultSer = env.serializer.newInstance()
val beforeSerialization = System.currentTimeMillis()
val valueBytes = resultSer.serialize(value)
val afterSerialization = System.currentTimeMillis()
// Note: accumulator updates must be collected after TaskMetrics is updated
val accumUpdates = task.collectAccumulatorUpdates()
val directResult = new DirectTaskResult(valueBytes, accumUpdates)
val serializedDirectResult = ser.serialize(directResult)
val resultSize = serializedDirectResult.limit
// directSend = sending directly back to the driver
val serializedResult: ByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
// 大於1G，直接丟棄
ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
} else if (resultSize > maxDirectResultSize) {
val blockId = TaskResultBlockId(taskId)
env.blockManager.putBytes(
blockId,
new ChunkedByteBuffer(serializedDirectResult.duplicate()),
StorageLevel.MEMORY_AND_DISK_SER)
// IndirectTaskResult間接結果
ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))
} else {
// 直接發送
serializedDirectResult
}
}
// CoraseGrainedExecutorBackend的statusUpdate方法
execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
} catch {...}