Spark发射任务源代码分析.docVIP

其中TaskSetManager类的resourceOffer()方法调用流程 CoarseGrainedSchedulerBackend的方法,注册executor，包括executorId,hostPort和cores.形成一个WorkOffer的列表，并发射任务 // Launch tasks returned by a set of resource offersprivate def launchTasks(tasks: Seq[Seq[TaskDescription]]) { for (task - tasks.flatten) { else { val executorData = executorDataMap(task.executorId) executorData.freeCores -= scheduler.CPUS_PER_TASK executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))} lancheTasks()方法的参数是TaskDescription的seq类型，对于其中的每个task，序列化，executorData.freeCores　-= 然后发射 CoarseGrainedSchedulerBackend的makeOffers()调用TaskSchedulerImpl的resourceOffers() 先对所有的WordOffer进行random Shuffle打乱顺序，并对TaskSet进行排序 然后不断 调用自己的resourceOfferSingleTaskSet()方法，直到taskSet发射成功。 private def resourceOfferSingleTaskSet( taskSet: TaskSetManager, maxLocality: TaskLocality, shuffledOffers: Seq[WorkerOffer], availableCpus: Array[Int], tasks: Seq[ArrayBuffer[TaskDescription]]) : Boolean = { var launchedTask = false for (i - 0 until shuffledOffers.size) { val execId = shuffledOffers(i).executorId val host = shuffledOffers(i).host if (availableCpus(i) = CPUS_PER_TASK) { try { for (task - taskSet.resourceOffer(execId, host, maxLocality)) { tasks(i) += task val tid = task.taskId taskIdToTaskSetManager(tid) = taskSet taskIdToExecutorId(tid) = execId executorsByHost(host) += execId availableCpus(i) -= CPUS_PER_TASK assert(availableCpus(i) = 0) launchedTask = true } } catch { case e: TaskNotSerializableException = logError(sResource offer failed, task set ${taskSet.name} was not serializable) // Do not offer resources for this task, but dont throw an error to allow other // task sets to be submitted. return launchedTask } } } return launchedTask} 在这个函数中分配execId.对于所有的WorkerOffer(executorId, host, cores), 如果可以获取的