碼上快樂

相關內容    簡體    繁體

Spark 源碼解析：TaskScheduler的任務提交和task最佳位置算法

本文轉載自   查看原文   2016-08-06 09:59   2293 
上篇文章 《  

Spark 源碼解析 : DAGScheduler中的DAG划分與提交

》 介紹了DAGScheduler的Stage划分算法。

本文繼續分析Stage被封裝成TaskSet，並將TaskSet提交到集群的Executor執行的過程

在DAGScheduler的submitStage方法中，將Stage划分完成，生成拓撲結構，當一個stage沒有父stage時候，會調用 DAGScheduler的 submitMissingTasks方法來提交該stage包含tasks。
首先來分析一下 DAGScheduler的 submitMissingTasks方法

1.獲取Task的最佳計算位置： 
   
   
   
           
    
    
    
    
            
  1.  val taskIdToLocations: Map[Int, Seq[TaskLocation]] = try {
    2. 
    
    
    
            
  2.  stage match {
    3. 
    
    
    
            
  3.  case s: ShuffleMapStage =>
    4. 
    
    
    
            
  4.  partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMap
    5. 
    
    
    
            
  5.  case s: ResultStage =>
    6. 
    
    
    
            
  6.  val job = s.activeJob.get
    7. 
    
    
    
            
  7.  partitionsToCompute.map { id =>
    8. 
    
    
    
            
  8.  val p = s.partitions(id)
    9. 
    
    
    
            
  9.  (id, getPreferredLocs(stage.rdd, p))
    10. 
    
    
    
            
  10.  }.toMap
    11. 
    
    
    
            
  11.  }
    12. 
    
    
    
            
  12.  }
    13.

核心是其中的getPreferredLocs方法，根據RDD的數據信息得到task的最佳計算位置，從而獲取較好的數據本地性。其中的細節這里先跳過，在以后的文章在做分析

2.序列化Task的Binary，並進行廣播。Executor端在執行task時會向反序列化Task。

3.根據stage的不同類型創建，為stage的每個分區創建創建task,並封裝成TaskSet。Stage分兩種類型ShuffleMapStage生成ShuffleMapTask，ResultStage生成ResultTask。 
    
    
    
            
    
     
     
     
             
  1.  val tasks: Seq[Task[_]] = try {
    2. 
     
     
     
             
  2.  stage match {
    3. 
     
     
     
             
  3.  case stage: ShuffleMapStage =>
    4. 
     
     
     
             
  4.  partitionsToCompute.map { id =>
    5. 
     
     
     
             
  5.  val locs = taskIdToLocations(id)
    6. 
     
     
     
             
  6.  val part = stage.rdd.partitions(id)
    7. 
     
     
     
             
  7.  new ShuffleMapTask(stage.id, stage.latestInfo.attemptId,
    8. 
     
     
     
             
  8.  taskBinary, part, locs, stage.internalAccumulators)
    9. 
     
     
     
             
  9.  }
    10. 
     
     
     
             
    11. 
     
     
     
             
  11.  case stage: ResultStage =>
    12. 
     
     
     
             
  12.  val job = stage.activeJob.get
    13. 
     
     
     
             
  13.  partitionsToCompute.map { id =>
    14. 
     
     
     
             
  14.  val p: Int = stage.partitions(id)
    15. 
     
     
     
             
  15.  val part = stage.rdd.partitions(p)
    16. 
     
     
     
             
  16.  val locs = taskIdToLocations(id)
    17. 
     
     
     
             
  17.  new ResultTask(stage.id, stage.latestInfo.attemptId,
    18. 
     
     
     
             
  18.  taskBinary, part, locs, id, stage.internalAccumulators)
    19. 
     
     
     
             
  19.  }
    20. 
     
     
     
             
  20.  }
    21.

4.調用TaskScheduler的submitTasks，提交TaskSet 
    
    
    
            
    
     
     
     
             
  1.  logInfo("Submitting " + tasks.size + " missing tasks from " + stage + " (" + stage.rdd + ")")
    2. 
     
     
     
             
  2.  stage.pendingPartitions ++= tasks.map(_.partitionId)
    3. 
     
     
     
             
  3.  logDebug("New pending partitions: " + stage.pendingPartitions)
    4. 
     
     
     
             
  4.  taskScheduler.submitTasks(new TaskSet(
    5. 
     
     
     
             
  5.  tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))
    6. 
     
     
     
             
  6.  stage.latestInfo.submissionTime = Some(clock.getTimeMillis())
    7.

submitTasks方法的實現在TaskScheduler的實現類TaskSchedulerImpl中。

4.1 TaskSchedulerImpl的submitTasks方法首先創建TaskSetManager。 
    
    
    
            
    
     
     
     
             
  1. val manager = createTaskSetManager(taskSet, maxTaskFailures)
    2. 
     
     
     
             
  2.  val stage = taskSet.stageId
    3. 
     
     
     
             
  3.  val stageTaskSets =
    4. 
     
     
     
             
  4.  taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
    5. 
     
     
     
             
  5.  stageTaskSets(taskSet.stageAttemptId) = manager
    6.

TaskSetManager負責管理 TaskSchedulerImpl中 一個單獨TaskSet，跟蹤每一個task，如果task失敗，負責重試task直到達到task重試次數的最多次數。並且通過延遲調度來執行task的位置感知調度。 

    
    
    
            
    
     
     
     
             
  1. private[spark] class TaskSetManager(
    2. 
     
     
     
             
  2.  sched: TaskSchedulerImpl,//綁定的TaskSchedulerImpl
    3. 
     
     
     
             
  3.  val taskSet: TaskSet,
    4. 
     
     
     
             
  4.  val maxTaskFailures: Int, //失敗最大重試次數
    5. 
     
     
     
             
  5.  clock: Clock = new SystemClock())
    6. 
     
     
     
             
  6.  extends Schedulable with Logging
    7.

4.2 將TaskSetManger加入schedulableBuilder 
    
    
    
            
    
     
     
     
             
  1. schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties) //將TaskSetManager加入rootPool調度池中，由schedulableBuilder決定調度順序
    2.

schedulableBuilder的類型是 SchedulerBuilder， SchedulerBuilder是一個trait，有兩個實現FIFO SchedulerBuilder和  Fair SchedulerBuilder，並且默認采用的是FIFO方式 

    
    
    
            
    
     
     
     
             
  1.  // default scheduler is FIFO
    2. 
     
     
     
             
  2.  private val schedulingModeConf = conf.get("spark.scheduler.mode", "FIFO")
    3.

而 schedulableBuilder的創建是在SparkContext創建SchedulerBackend和TaskScheduler后調用TaskSchedulerImpl的初始化方法進行創建的。 

    
    
    
            
    
     
     
     
             
  1.  def initialize(backend: SchedulerBackend) {
    2. 
     
     
     
             
  2.  this.backend = backend
    3. 
     
     
     
             
  3.  // temporarily set rootPool name to empty
    4. 
     
     
     
             
  4.  rootPool = new Pool("", schedulingMode, 0, 0)
    5. 
     
     
     
             
  5.  schedulableBuilder = {
    6. 
     
     
     
             
  6.  schedulingMode match {
    7. 
     
     
     
             
  7.  case SchedulingMode.FIFO =>
    8. 
     
     
     
             
  8.  new FIFOSchedulableBuilder(rootPool)
    9. 
     
     
     
             
  9.  case SchedulingMode.FAIR =>
    10. 
     
     
     
             
  10.  new FairSchedulableBuilder(rootPool, conf)
    11. 
     
     
     
             
  11.  }
    12. 
     
     
     
             
  12.  }
    13. 
     
     
     
             
  13.  schedulableBuilder.buildPools()
    14. 
     
     
     
             
  14.  }
    15.

schedulableBuilder是TaskScheduler中一個重要成員，他根據調度策略決定了TaskSetManager的調度順序。

4.3 接下來調用SchedulerBackend的riviveOffers方法對Task進行調度，決定task具體運行在哪個Executor中。

調用CoarseGrainedSchedulerBackend的riviveOffers方法，該方法給driverEndpoint發送ReviveOffer消息 

   
   
   
           
    
    
    
    
            
  1.  override def reviveOffers() {
    2. 
    
    
    
            
  2.  driverEndpoint.send(ReviveOffers)
    3. 
    
    
    
            
  3.  }
    4.

driverEndpoint收到 ReviveOffer消息后調用makeOffers方法 
    
    
    
            
    
     
     
     
             
  1.  // Make fake resource offers on all executors
    2. 
     
     
     
             
  2.  private def makeOffers() {
    3. 
     
     
     
             
  3.  //過濾出活躍狀態的Executor
    4. 
     
     
     
             
  4.  val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
    5. 
     
     
     
             
  5.       //將Executor封裝成WorkerOffer對象
    6. 
     
     
     
             
  6.  val workOffers = activeExecutors.map { case (id, executorData) =>
    7. 
     
     
     
             
  7.  new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
    8. 
     
     
     
             
  8.  }.toSeq
    9. 
     
     
     
             

    10. 
     
     
     
             
  10.   launchTasks(scheduler.resourceOffers(workOffers))
    11. 
     
     
     
             
  11.  }
    12.

注意：上面代碼中的executorDataMap，在客戶的向Master注冊Application的時候，Master已經為 Application分配並啟動好Executor，然后注冊給 CoarseGrainedSchedulerBackend，注冊信息就是存儲在executorDataMap數據結構中。

准備好計算資源后，接下來TaskSchedulerImpl基於這些計算資源為task分配Executor。
我們看一下 TaskSchedulerImpl的resourceOffers方法： 

    
    
    
            
    
     
     
     
             
  1.  // 隨機打亂offers
    2. 
     
     
     
             
  2.  val shuffledOffers = Random.shuffle(offers)
    3. 
     
     
     
             

    4. 
     
     
     
             
  4.   // 構建一個二維數組，保存每個Executor上將要分配的那些task
    5. 
     
     
     
             
  5.  val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
    6. 
     
     
     
             
  6.  val availableCpus = shuffledOffers.map(o => o.cores).toArray
    7. 
    
    
    
            

    
    
    
            
    
     
     
     
             
  1.   
    2. 
     
     
     
             
  2.     //根據SchedulerBuilder的調度算法，給TaskManager排好序
    3. 
     
     
     
             
       
              
    1. val sortedTaskSets = rootPool.getSortedTaskSetQueue
      2.  
             
    
     
     
     
             
  3.  for (taskSet <- sortedTaskSets) {
    4. 
     
     
     
             
  4.  logDebug("parentName: %s, name: %s, runningTasks: %s".format(
    5. 
     
     
     
             
  5.  taskSet.parent.name, taskSet.name, taskSet.runningTasks))
    6. 
     
     
     
             
  6.  if (newExecAvail) {
    7. 
     
     
     
             
  7.  taskSet.executorAdded()
    8. 
     
     
     
             
  8.  }
    9. 
     
     
     
             
  9.  }
    10. 
     
     
     
             
    11. 
     
     
     
             
  11.   // 使用雙重循環，對每一個taskset 依照調度的順序，依次按照本地性級別順序嘗試啟動task 
    12. 
     
     
     
             
  12.  // 數據本地性級別順序: PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY
    13. 
     
     
     
             
  13.  var launchedTask = false
    14. 
     
     
     
             
  14.  for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
    15. 
     
     
     
             
  15.  do {
    16. 
     
     
     
             
  16.  launchedTask = resourceOfferSingleTaskSet(
    17. 
     
     
     
             
  17.  taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
    18. 
     
     
     
             
  18.  } while (launchedTask)
    19. 
     
     
     
             
  19.  }
    20. 
     
     
     
             
    21. 
     
     
     
             
  21.  if (tasks.size > 0) {
    22. 
     
     
     
             
  22.  hasLaunchedTask = true
    23. 
     
     
     
             
  23.  }
    24. 
     
     
     
             
  24.  return tasks
    25.

下面看看  resourceOfferSingleTaskSet 方法： 
用當前的數據本地性，調用TaskSetManager的resourceOffer方法，在當前executor上分配task
    
    
    
            
    
     
     
     
             
  1. private def resourceOfferSingleTaskSet(
    2. 
     
     
     
             
  2.  taskSet: TaskSetManager,
    3. 
     
     
     
             
  3.  maxLocality: TaskLocality,
    4. 
     
     
     
             
  4.  shuffledOffers: Seq[WorkerOffer],
    5. 
     
     
     
             
  5.  availableCpus: Array[Int],
    6. 
     
     
     
             
  6.  tasks: Seq[ArrayBuffer[TaskDescription]]) : Boolean = {
    7. 
     
     
     
             
  7.  var launchedTask = false
    8. 
     
     
     
             
  8.  for (i <- 0 until shuffledOffers.size) {
    9. 
     
     
     
             
  9.  val execId = shuffledOffers(i).executorId
    10. 
     
     
     
             
  10.  val host = shuffledOffers(i).host
    11. 
     
     
     
             
  11.         //如果executor 的cup數大於 每個task的cup數目（值為1）
    12. 
     
     
     
             
  12.  if (availableCpus(i) >= CPUS_PER_TASK) {
    13. 
     
     
     
             
  13.  try {
    14. 
     
     
     
             
  14.         //
    15. 
     
     
     
             
  15.  for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
    16. 
     
     
     
             
  16.  tasks(i) += task
    17. 
     
     
     
             
  17.  val tid = task.taskId
    18. 
     
     
     
             
  18.  taskIdToTaskSetManager(tid) = taskSet
    19. 
     
     
     
             
  19.  taskIdToExecutorId(tid) = execId
    20. 
     
     
     
             
  20.  executorIdToTaskCount(execId) += 1
    21. 
     
     
     
             
  21.  executorsByHost(host) += execId
    22. 
     
     
     
             
  22.  availableCpus(i) -= CPUS_PER_TASK
    23. 
     
     
     
             
  23.  assert(availableCpus(i) >= 0)
    24. 
     
     
     
             
  24.  launchedTask = true
    25. 
     
     
     
             
  25.  }
    26. 
     
     
     
             
  26.  }
    27.

為Task分配好資源之后，DriverEndpint調用launchTask方法將task在Executor上啟動運行。task在Executor上的啟動運行過程，在后面的文章中會繼續分析，敬請關注。

總結一下調用過程：
TaskSchedulerImpl# submitTasks
CoarseGrainedSchedulerBackend# riviveOffers
CoarseGrainedSchedulerBackend$DriverEndpoint#makeOffers
  |-TaskSchedulerImpl#resourceOffers(offers) 為offers分配task 
    |-  TaskSchedulerImpl # resourceOfferSingleTaskSet
CoarseGrainedSchedulerBackend$DriverEndpoint#launchTask










免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 spark 源碼分析之四 -- TaskScheduler的創建和啟動過程 提交任務到Spark Spark提交任務到集群 Spark 源碼解析 : DAGScheduler中的DAG划分與提交 提交任務到spark（以wordcount為例） Spark：使用Java代碼提交spark任務 Flink源碼解析（四）——從Flink集群部署和任務提交模式看Flink任務的核心組件 Spark 源碼分析 -- task實際執行過程 spark任務提交到yarn上命令總結 spark提交任務的兩種的方法
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM