Spark Executor 執行Task的過程

前言

前面分析了task scheduler劃分task的過程，task在task schedulerimpl類中被包裝成stagetasksets，然後由driverendpoint傳送，最後由coarsegrainedschedulerbackend序列化並啟動executor。現在開始分析executor執行任務的過程。正文

coarsegrainedschedulerbackend傳送一條lunchtask訊息後(executordata.executorendpoint.send(launchtask(new serializablebuffer(serializedtask)))),executor收到訊息，開始執行lunchtask方法。

case
launchtask
(data)
=>
if(executor == null)
else

executor.launchtask(this, taskdesc)把任務詳情也傳入進去，然後啟動乙個taskrunner，並把taskdesc也傳入進去：val tr = new taskrunner(context, taskdescription)，這個taskrunner會把task反序列化出來，並且這個run()方法會在獲取結果時執行task.run()方法。

task = ser.deserialize[task[any]](
taskdescription.serializedtask, thread.currentthread.getcontextclassloader)
task.localproperties = taskdescription.properties
task.
settaskmemorymanager
(taskmemorymanager)
## 執行task.
run(
)方法 val res = task.
run(
taskattemptid = taskid,
attemptnumber = taskdescription.attemptnumber,
metricssystem = env.metricssystem)
threwexception =
false
res

執行task.run()方法，會執行runtask()方法，這個runtask()方法沒有實現，因為task有兩類，一類是shufflemaptask，一類是resulttask。所以需要各自實現runtask()方法。我們先看一下shufflemaptask的runtask()方法。

override def runtask
(context: taskcontext)
: mapstatus =
else
0l val ser = sparkenv.get.closureserializer.
newinstance()
val (rdd, dep)
= ser.deserialize[
(rdd[_]
, shuffledependency[_, _, _])]
( bytebuffer.
wrap
(taskbinary.value)
, thread.currentthread.getcontextclassloader)
_executordeserializetime = system.
currenttimemillis()
- deserializestarttime
_executordeserializecputime =
if(threadmxbean.iscurrentthreadcputimesupported)
else
0l var writer: shufflewriter[any, any]
= null
trycatch
}catch
throw e
}}

val (rdd, dep) = ser.deserialize[(rdd[_], shuffledependency這條**就是從反序列化的資料裡拿到stage的最後乙個rdd和它的依賴，並通過sparkenv建立了乙個manager，然後通過manager建立了乙個shuffle writer，這個shuffle writer就是將任務的計算結果寫到本地磁碟的角色，所以它很重要，shuffle writer也有3種，這裡不詳細講。此外，還有shuffle reader負責拉取shuffle writer寫入的資料。這個以後會詳細寫。writer呼叫write()方法將資料迭代地寫入磁碟。總結

executor 執行task的過程寫的比較簡單，但其實裡面涉及到比較多的知識點，比如sparkenv，shuffle writer等等，接下來會分別介紹這些知識點。

Spark Executor 執行Task的過程

spark Executor啟動過程分析

執行完畢再往下執行？

同步執行和非同步執行

Spark Executor 執行Task的過程

spark Executor啟動過程分析

執行完畢再往下執行？

同步執行和非同步執行

相關推薦