228
spark源碼分析之任務調度篇
DAG的生成
概述
DAG(Directed Acyclic Graph)叫做有向無環圖，原始的RDD通過一係列的轉換就就形成了DAG，根據RDD之間的依賴關係的不同將DAG劃分成不同的Stage，對於窄依賴，partition的轉換處理在Stage中完成計算。對於寬依賴，由於有Shuffle的存在，隻能在parent RDD處理完成後，才能開始接下來的計算，因此寬依賴是劃分Stage的依據。
窄依賴 指的是每一個父RDD的Partition最多被子RDD的一個Partition使用
寬依賴 指的是多個子RDD的Partition會依賴同一個父RDD的Partition
DAGScheduler調度隊列
當我們看完Executor的創建與啟動流程後，我們繼續在SparkContext的構造方法中繼續查看
class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationClient {
。。。。。。
private[spark] def createSparkEnv(
conf: SparkConf,
isLocal: Boolean,
listenerBus: LiveListenerBus): SparkEnv = {
//通過SparkEnv來創建createDriverEnv
SparkEnv.createDriverEnv(conf, isLocal, listenerBus)
}
//在這裏調用了createSparkEnv，返回一個SparkEnv對象，這個對象裏麵有很多重要屬性，最重要的ActorSystem
private[spark] val env = createSparkEnv(conf, isLocal, listenerBus)
SparkEnv.set(env)
//創建taskScheduler
// Create and start the scheduler
private[spark] var (schedulerBackend, taskScheduler) =
SparkContext.createTaskScheduler(this, master)
//創建DAGScheduler
dagScheduler = new DAGScheduler(this)
//啟動TaksScheduler
taskScheduler.start()
。。。。。
}
在構造方法中還創建了一個DAGScheduler對象，這個類的任務就是用來劃分Stage任務的，構造方法中初始化了 private[scheduler] val eventProcessLoop = new DAGSchedulerEventProcessLoop(this)
DAGSchedulerEventProcessLoop是一個事件總線對象，用來負責任務的分發，在構造方法eventProcessLoop.start()被調用，該方法是父類EventLoop的start
def start(): Unit = {
if (stopped.get) {
throw new IllegalStateException(name + " has already been stopped")
}
// Call onStart before starting the event thread to make sure it happens before onReceive
onStart()
eventThread.start()
}
調用了eventThread的start方法，開啟了一個線程
private val eventThread = new Thread(name) {
setDaemon(true)
override def run(): Unit = {
try {
while (!stopped.get) {
val event = eventQueue.take()
try {
onReceive(event)
} catch {
case NonFatal(e) => {
try {
onError(e)
} catch {
case NonFatal(e) => logError("Unexpected error in " + name, e)
}
}
}
}
} catch {
case ie: InterruptedException => // exit even if eventQueue is not empty
case NonFatal(e) => logError("Unexpected error in " + name, e)
}
}
}
run方法中不斷的從LinkedBlockingDeque阻塞隊列中取消息，然後調用onReceive(event)方法，該方法是由子類DAGSchedulerEventProcessLoop實現的
override def onReceive(event: DAGSchedulerEvent): Unit = event match {
case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties) =>
//調用dagScheduler來出來提交任務
dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite,
listener, properties)
case StageCancelled(stageId) =>
dagScheduler.handleStageCancellation(stageId)
case JobCancelled(jobId) =>
dagScheduler.handleJobCancellation(jobId)
case JobGroupCancelled(groupId) =>
dagScheduler.handleJobGroupCancelled(groupId)
case AllJobsCancelled =>
dagScheduler.doCancelAllJobs()
case ExecutorAdded(execId, host) =>
dagScheduler.handleExecutorAdded(execId, host)
case ExecutorLost(execId) =>
dagScheduler.handleExecutorLost(execId, fetchFailed = false)
case BeginEvent(task, taskInfo) =>
dagScheduler.handleBeginEvent(task, taskInfo)
case GettingResultEvent(taskInfo) =>
dagScheduler.handleGetTaskResult(taskInfo)
case completion @ CompletionEvent(task, reason, _, _, taskInfo, taskMetrics) =>
dagScheduler.handleTaskCompletion(completion)
case TaskSetFailed(taskSet, reason) =>
dagScheduler.handleTaskSetFailed(taskSet, reason)
case ResubmitFailedStages =>
dagScheduler.resubmitFailedStages()
}
onReceive中會匹配到傳入的任務類型，執行相應的邏輯。到此DAGScheduler的調度隊列會一直掛起，不斷輪詢隊列中的任務。
DAG提交Task任務流程
當RDD經過一係列的轉換Transformation方法後，最終要執行Action動作方法，這裏比如WordCount程序中最後調用collect()方法時會將數據提交到Master上運行，任務真正的被執行，這裏的方法執行過程如下
/**
* Return an array that contains all of the elements in this RDD.
*/
def collect(): Array[T] = {
val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
Array.concat(results: _*)
}
sc 是SparkContext對象，這裏調用 一個runJob該方法調用多次重載的方法後,該方法最終會調用 dagScheduler.runJob
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
allowLocal: Boolean,
resultHandler: (Int, U) => Unit) {
if (stopped) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
if (conf.getBoolean("spark.logLineage", false)) {
logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
}
//dagScheduler出現了，可以切分stage
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal,
resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
rdd.doCheckpoint()
}
dagScheduler的runJob是我們比較關心的
def runJob[T, U: ClassTag](
。。。。。
val waiter = submitJob(rdd, func, partitions, callSite, allowLocal, resultHandler, properties)
waiter.awaitResult() match {
case JobSucceeded => {
logInfo("Job %d finished: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
}
case JobFailed(exception: Exception) =>
logInfo("Job %d failed: %s, took %f s".format
(waiter.jobId, callSite.shortForm, (System.nanoTime - start) / 1e9))
throw exception
}
}
這裏麵的我們主要看的是submitJob(rdd, func, partitions, callSite, allowLocal, resultHandler, properties)提交任務
def submitJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
allowLocal: Boolean,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
。。。。。。
//把job加入到任務隊列裏麵
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter, properties))
waiter
}
這裏比較關鍵的地方是eventProcessLoop.post往任務隊列中加入一個JobSubmitted類型的任務，eventProcessLoop是在構造方法中就初始化好的事件總線對象，內部有一個線程不斷的輪詢隊列裏的任務
輪詢到任務後調用onReceive方法匹配任務類型，在這裏我們提交的任務是JobSubmitted類型
case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties) =>
//調用dagScheduler來出來提交任務
dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite,
listener, properties)
調用了handleJobSubmitted方法，接下來查看該方法
private[scheduler] def handleJobSubmitted(jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
allowLocal: Boolean,
callSite: CallSite,
listener: JobListener,
properties: Properties) {
var finalStage: Stage = null
try {
// New stage creation may throw an exception if, for example, jobs are run on a
// HadoopRDD whose underlying HDFS files have been deleted.
//最終的stage
finalStage = newStage(finalRDD, partitions.size, None, jobId, callSite)
} catch {
case e: Exception =>
logWarning("Creating new stage failed due to exception - job: " + jobId, e)
listener.jobFailed(e)
return
}
。。。。
submitStage(finalStage)
}
上麵的代碼中，調用了newStage進行任務的劃分，該方法是劃分任務的核心方法，劃分任務的根據最後一個依賴關係作為開始，通過遞歸，將每個寬依賴做為切分Stage的依據，切分Stage的過程是流程中的一環，但在這裏不詳細闡述，當任務切分完畢後，代碼繼續執行來到submitStage(finalStage)這裏開始進行任務提交
這裏以遞歸的方式進行任務的提交
//遞歸的方式提交stage
private def submitStage(stage: Stage) {
val jobId = activeJobForStage(stage)
if (jobId.isDefined) {
logDebug("submitStage(" + stage + ")")
if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
val missing = getMissingParentStages(stage).sortBy(_.id)
logDebug("missing: " + missing)
if (missing == Nil) {
logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
//提交任務
submitMissingTasks(stage, jobId.get)
} else {
for (parent <- missing) {
submitStage(parent)
}
waitingStages += stage
}
}
} else {
abortStage(stage, "No active job for stage " + stage.id)
}
}
調用submitMissingTasks(stage, jobId.get)提交任務，將每一個Stage和jobId傳入
private def submitMissingTasks(stage: Stage, jobId: Int) {
。。。。。
if (tasks.size > 0) {
logInfo("Submitting " + tasks.size + " missing tasks from " + stage + " (" + stage.rdd + ")")
stage.pendingTasks ++= tasks
logDebug("New pending tasks: " + stage.pendingTasks)
//taskScheduler提交task
taskScheduler.submitTasks(
new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.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)
logDebug("Stage " + stage + " is actually done; %b %d %d".format(
stage.isAvailable, stage.numAvailableOutputs, stage.numPartitions))
}
}
這裏的代碼我們需要關注的是taskScheduler.submitTasks(
new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.jobId, properties))
創建了一個TaskSet對象，將所有任務的信息封裝，包括task任務列表，stageId,任務id,分區數參數等
Task任務調度
override def submitTasks(taskSet: TaskSet) {
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
//創建TaskSetManager保存了taskSet任務列表
val manager = createTaskSetManager(taskSet, maxTaskFailures)
activeTaskSets(taskSet.id) = manager
//將任務加入調度池
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
if (!isLocal && !hasReceivedTask) {
starvationTimer.scheduleAtFixedRate(new TimerTask() {
override def run() {
if (!hasLaunchedTask) {
logWarning("Initial job has not accepted any resources; " +
"check your cluster UI to ensure that workers are registered " +
"and have sufficient resources")
} else {
this.cancel()
}
}
}, STARVATION_TIMEOUT, STARVATION_TIMEOUT)
}
hasReceivedTask = true
}
//接受任務
backend.reviveOffers()
}
該方法比較重要，主要將任務加入調度池，最後調用了backend.reviveOffers()這裏的backend是CoarseGrainedSchedulerBackend一個Executor任務調度對象
override def reviveOffers() {
//自己給自己發消息
driverActor ! ReviveOffers
}
這裏用了內部的DriverActor對象發送了一個內部消息給自己，接下來查看receiver方法接受的消息
case ReviveOffers =>
makeOffers()
收到消息後調用了makeOffers()方法
def makeOffers() {
launchTasks(scheduler.resourceOffers(executorDataMap.map { case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
}.toSeq))
}
makeOffers方法中，將Executor的信息集合與調度池中的Tasks封裝成WokerOffers列表傳給了launchTasks
def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
。。。。。。
//把task序列化
val serializedTask = ser.serialize(task)
。。。。。
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
//把序列化好的task發送給Executor
executorData.executorActor ! LaunchTask(new SerializableBuffer(serializedTask))
}
}
}
launchTasks方法將遍曆Tasks集合,每個Task任務序列化，發送啟動Task執行消息的給Executor
Executor的onReceive方法
//DriverActor發送給Executor的啟動Task的消息
case LaunchTask(data) =>
if (executor == null) {
logError("Received LaunchTask command but executor was null")
System.exit(1)
} else {
val ser = env.closureSerializer.newInstance()
//把Task反序列化
val taskDesc = ser.deserialize[TaskDescription](data.value)
logInfo("Got assigned task " + taskDesc.taskId)
//啟動task
executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,
taskDesc.name, taskDesc.serializedTask)
}
Executor收到DriverActor發送的啟動Task的消息，這裏才開始真正執行任務了，將收到的Task序列化信息反序列化,調用Executor的launchTask方法執行任務
def launchTask(
context: ExecutorBackend,
taskId: Long,
attemptNumber: Int,
taskName: String,
serializedTask: ByteBuffer) {
//把task的描述信息放到了一份TaskRunner
val tr = new TaskRunner(context, taskId = taskId, attemptNumber = attemptNumber, taskName,
serializedTask)
runningTasks.put(taskId, tr)
//然後把TaskRunner丟到線程池裏麵
threadPool.execute(tr)
}
launchTask內將Task提交到線程池去運行，TaskRunner是Runnable對象，裏麵的run方法執行了我們app生成的每一個RDD的鏈上的邏輯。
最後更新：2017-05-01 08:01:17