您的位置: 首页  >  文章  >  大数据：Spark Core（二）Driver上的Task的生成、分配、调度

大数据：Spark Core（二）Driver上的Task的生成、分配、调度

分类: 文章 • 2025-04-17 13:25:52

1. 什么是Task?

在前面的章节里描述过几个角色，Driver(Client),Master,Worker(Executor)，Driver会提交Application到Master进行Worker上的Executor上的调度，显然这些都不是Task.

Spark上的几个关系可以这样理解：

  • Application: Application是Driver在构建SparkContent的上下文的时候创建的，就像申报员，现在要构建一个能完成任务的集群，需要申报的是这次需要多少个Executor(可以简单理解为虚拟的机器)，每个Executor需要多少CPU,多少内存。
  • Job: 这是Driver在调用Action的时候生成的Job，让DAGScheduler线程进行最后的调度，代表着这时候RDD的状态分析完了，需要进行最后的Stage分析了，Job并不是提交给Executor运行的，一个Application存在多个Job
  • Task: 一个Job可以分成多个Task, 相当于这些Task分到了一个Group里，这个Group的ID就是Job ID

2. Task的类型

Task的类型和Stage相关，关于Stage，以及Stage之间的相关依赖构成任务的不同提交，就不在这篇描述了

ShuffleMapStage 转化成 ShuffleMapTask

ResultStage 转化成为 ResultTask

当Spark上的action算子，通过DAG进行提交任务的时候，会通过Stage来决定提交什么类型的任务，具体的实现都在DAGScheduler.scala 的submitMissingTasks方法中。

3. 同一个Stage的Task数量

Spark是一个分布式的执行任务的框架，那么同一个Stage的并行任务的拆分就非常的重要，在任务的分解中并不只是stage的步骤的分解，同时也是对同一个Stage中的要分析的数据分解，而对数据的分解直接决定对同一个Stage所提交的任务的数量。对Stage的任务拆解决定着任务的之间的关系，而对同一个Stage的分析数据进行拆解控制着任务的数量。

比如基于拆解的分析数据的而执行的算子象map，这些任务都是独立的，并没有对数据进行最后的归并和整理，这些task是完全可以进行并行计算的，对同一个Stage的task的数量在Spark上是可以控制的。

在这里以ParallelCollectionRDD为简单的例子，先看DAGScheduler.submitMissingTasks的方法

[java] view plain copy
 print?
  1. private def submitMissingTasks(stage: Stage, jobId: Int) {  
  2.    logDebug("submitMissingTasks(" + stage + ")")  
  3.    // Get our pending tasks and remember them in our pendingTasks entry  
  4.    stage.pendingPartitions.clear()  
  5.   
  6.    // First figure out the indexes of partition ids to compute.  
  7.    val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()  
  8.    。。。。。。。。。。。  
  9.    val tasks: Seq[Task[_]] = try {  
  10.      stage match {  
  11.        case stage: ShuffleMapStage =>  
  12.          partitionsToCompute.map { id =>  
  13.            val locs = taskIdToLocations(id)  
  14.            val part = stage.rdd.partitions(id)  
  15.            new ShuffleMapTask(stage.id, stage.latestInfo.attemptId,  
  16.              taskBinary, part, locs, stage.latestInfo.taskMetrics, properties, Option(jobId),  
  17.              Option(sc.applicationId), sc.applicationAttemptId)  
  18.          }  
  19.   
  20.        case stage: ResultStage =>  
  21.          partitionsToCompute.map { id =>  
  22.            val p: Int = stage.partitions(id)  
  23.            val part = stage.rdd.partitions(p)  
  24.            val locs = taskIdToLocations(id)  
  25.            new ResultTask(stage.id, stage.latestInfo.attemptId,  
  26.              taskBinary, part, locs, id, properties, stage.latestInfo.taskMetrics,  
  27.              Option(jobId), Option(sc.applicationId), sc.applicationAttemptId)  
  28.          }  
  29.      }  
  30.    } catch {  
  31.      case NonFatal(e) =>  
  32.        abortStage(stage, s"Task creation failed: $e\n${Utils.exceptionString(e)}", Some(e))  
  33.        runningStages -= stage  
  34.        return  
  35.    }  

生产task的数量是由val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()来决定的，在ShuffleMapStage里

[java] view plain copy
 print?
  1. override def findMissingPartitions(): Seq[Int] = {  
  2.     val missing = (0 until numPartitions).filter(id => outputLocs(id).isEmpty)  
  3.     assert(missing.size == numPartitions - _numAvailableOutputs,  
  4.       s"${missing.size} missing, expected ${numPartitions - _numAvailableOutputs}")  
  5.     missing  
  6.   }  

可以看到具体是由numPartitions来决定的，在来看numPartitions

[java] view plain copy
 print?
  1. val numPartitions = rdd.partitions.length  
由rdd.partitions来决定的,对ShuffleMapStage来说rdd就是最后一个value类型的transformation 的RDD，比如常见的MapPartitionsRDD

在MapPartitionsRDD来说的partitions

[java] view plain copy
 print?
  1. override def getPartitions: Array[Partition] = firstParent[T].partitions  
是transformation的算子链中的第一个，我们以ParallelCollectionRDD为例子，比如常见的对应的例子:

[java] view plain copy
 print?
  1. sparkcontext.parallelize(exampleApacheLogs)  
在ParallelCollectionRDD中

[java] view plain copy
 print?
  1. override def getPartitions: Array[Partition] = {  
  2.     val slices = ParallelCollectionRDD.slice(data, numSlices).toArray  
  3.     slices.indices.map(i => new ParallelCollectionPartition(id, i, slices(i))).toArray  
  4.   }  
在ParallelCollectionRDD中数据的Partitions是由numSlices来决定的

[java] view plain copy
 print?
  1. def parallelize[T: ClassTag](  
  2.     seq: Seq[T],  
  3.     numSlices: Int = defaultParallelism): RDD[T] = withScope {  
  4.   assertNotStopped()  
  5.   new ParallelCollectionRDD[T](this, seq, numSlices, Map[Int, Seq[String]]())  
  6. }  
numSlices 是可以在parallelize函数中传入，而默认使用defaultParallelism的参数控制

[java] view plain copy
 print?
  1. def defaultParallelism: Int = {  
  2.     assertNotStopped()  
  3.     taskScheduler.defaultParallelism  
  4.   }  
  5. override def defaultParallelism(): Int = backend.defaultParallelism()  

在CoarseGrainedSchedulerBackend.scala 的类中:

[java] view plain copy
 print?
  1. override def defaultParallelism(): Int = {  
  2.   conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))  
  3. }  

默认的值是受以下控制：

  1. 配置文件spark.default.parallelism
  2. totalCoreCount 的值： CoarseGrainedSchedulerBackend是一个executor管理的backend，里面维护着executor的信息，totalCoreCount就是executor汇报上来的核数，注意因为executor汇报自己是在application分配好后发生的，汇报的信息和获取totalCoreCount的线程是异步的，也就是如果executor没有汇报上来，totalCoreCount.get（）的值并不准确（根绝Master对executor的分配策略，是无法保证分配多少个executor, 在这里spark更依赖executor主动的向driver汇报），这里的策略是无法保证准确的获取executor的核数。
  3. 如果没有设置spark.default.parallelism，最小值是2

依赖于rdd.partitions的策略，最后决定task的分配数量。

4. Task的提交和调度分配

大数据：Spark Core（二）Driver上的Task的生成、分配、调度


在本篇中主要描述集群下的任务调度

4.1 Task的提交

在DAGScheduler将一个Stage中所分配的Task形成一个TaskSet进行提交，在TaskSet里所保存的是Task的集合，还有Stage的Id，以及JobId，注意在这里JobId是作为一个优先级的参数，作为后序队列调度的参数。

在TaskSchedulerImpl.scala中

[java] view plain copy
 print?
  1. override def submitTasks(taskSet: TaskSet) {  
  2.   val tasks = taskSet.tasks  
  3.   logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")  
  4.   this.synchronized {  
  5.     val manager = createTaskSetManager(taskSet, maxTaskFailures)  
  6.     val stage = taskSet.stageId  
  7.     val stageTaskSets =  
  8.       taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])  
  9.     stageTaskSets(taskSet.stageAttemptId) = manager  
  10.     val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>  
  11.       ts.taskSet != taskSet && !ts.isZombie  
  12.     }  
  13.     if (conflictingTaskSet) {  
  14.       throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +  
  15.         s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")  
  16.     }  
  17.     schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)  
  18.   
  19.     if (!isLocal && !hasReceivedTask) {  
  20.       starvationTimer.scheduleAtFixedRate(new TimerTask() {  
  21.         override def run() {  
  22.           if (!hasLaunchedTask) {  
  23.             logWarning("Initial job has not accepted any resources; " +  
  24.               "check your cluster UI to ensure that workers are registered " +  
  25.               "and have sufficient resources")  
  26.           } else {  
  27.             this.cancel()  
  28.           }  
  29.         }  
  30.       }, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)  
  31.     }  
  32.     hasReceivedTask = true  
  33.   }  
  34.   backend.reviveOffers()  
  35. }  
将TaskSet 封装成TaskSetManager，通过schedulableBuilder去添加TaskSetManager到队列中，在Spark中，有两种形态

大数据：Spark Core（二）Driver上的Task的生成、分配、调度

  1. FIFOSchedulableBuilder: 单一pool
  2. FairSchedulableBuilder:   多个pool

4.1.1 FairSchedulableBuilder pool池

通过fairsscheduler.xml的模版来设置参数来控制pool的调度

[html] view plain copy
 print?
  1. <allocations>  
  2.   <pool name="production1">  
  3.     <schedulingMode>FAIR</schedulingMode>  
  4.     <weight>3</weight>  
  5.     <minShare>4</minShare>  
  6.   </pool>  
  7.   <pool name="<span style="font-family: Arial, Helvetica, sans-serif;">production2</span><span style="font-family: Arial, Helvetica, sans-serif;">"></span>  
  8.     <schedulingMode><span style="color: rgb(51, 51, 51); font-family: 'Source Code Pro', monospace; font-size: 14px; font-variant-ligatures: normal; orphans: 2; widows: 2; background-color: rgba(128, 128, 128, 0.0470588);">FAIR</span></schedulingMode>  
  9.     <weight>5</weight>  
  10.     <minShare>2</minShare>  
  11.   </pool>  
  12. </allocations>  

参数的定义：

  • name:   调度池的名称，可根据该参数使用指定pool，EX: sc.setLocalProperty("spark.scheduler.pool", "production1") 
  • weight:  调度池的权重，调度池根据该参数分配资源。
  • minShare: 调度池需要的最小资源数（CPU核数），公平调度器首先会尝试为每个调度池分配最少minShare资源，然后剩余资源才会按照weight大小继续分配
  • schedulingMode: 调度池内的调度模式

在TaskSchedulerImpl在submitTasks添加TaskSetManager到pool后，调用了backend.reviveOffers();

[java] view plain copy
 print?
  1. override def reviveOffers() {  
  2.   driverEndpoint.send(ReviveOffers)  
  3. }  

[plain] view plain copy
 print?
  1.   
是向driver的endpoint发送了reviveoffers的消息，Spark中的许多操作都是通过消息来传递的，哪怕DAGScheduler的线程和endpoint的线程都是同一个Driver进程

4.2 Task的分配

Netty 的dispatcher线程接受到revievoffers的消息后，CoarseGrainedSchedulerBackend

[plain] view plain copy
 print?
  1. case ReviveOffers =>  
  2.   makeOffers()  

调用了makeoffers函数

[plain] view plain copy
 print?
  1. private def makeOffers() {  
  2.       // Filter out executors under killing  
  3.       val activeExecutors = executorDataMap.filterKeys(executorIsAlive)  
  4.       val workOffers = activeExecutors.map { case (id, executorData) =>  
  5.         new WorkerOffer(id, executorData.executorHost, executorData.freeCores)  
  6.       }.toIndexedSeq  
  7.       launchTasks(scheduler.resourceOffers(workOffers))  
  8.     }  

makeOffers里进行了资源的调度，netty中接收所有的信息，同时也在CoarseGrainedSchedulerBackend中维护着executor的状态map:executorDataMap，executor的状态是executor主动汇报的。

通过scheduler.resourceOffers来进行task的资源分配到executor中

[java] view plain copy
 print?
  1. def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {  
  2.    // Mark each slave as alive and remember its hostname  
  3.    // Also track if new executor is added  
  4.    var newExecAvail = false  
  5.    for (o <- offers) {  
  6.      if (!hostToExecutors.contains(o.host)) {  
  7.        hostToExecutors(o.host) = new HashSet[String]()  
  8.      }  
  9.      if (!executorIdToRunningTaskIds.contains(o.executorId)) {  
  10.        hostToExecutors(o.host) += o.executorId  
  11.        executorAdded(o.executorId, o.host)  
  12.        executorIdToHost(o.executorId) = o.host  
  13.        executorIdToRunningTaskIds(o.executorId) = HashSet[Long]()  
  14.        newExecAvail = true  
  15.      }  
  16.      for (rack <- getRackForHost(o.host)) {  
  17.        hostsByRack.getOrElseUpdate(rack, new HashSet[String]()) += o.host  
  18.      }  
  19.    }  
  20.   
  21.    // Randomly shuffle offers to avoid always placing tasks on the same set of workers.  
  22.    val shuffledOffers = Random.shuffle(offers)  
  23.    // Build a list of tasks to assign to each worker.  
  24.    val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))  
  25.    val availableCpus = shuffledOffers.map(o => o.cores).toArray  
  26.    val sortedTaskSets = rootPool.getSortedTaskSetQueue  
  27.    for (taskSet <- sortedTaskSets) {  
  28.      logDebug("parentName: %s, name: %s, runningTasks: %s".format(  
  29.        taskSet.parent.name, taskSet.name, taskSet.runningTasks))  
  30.      if (newExecAvail) {  
  31.        taskSet.executorAdded()  
  32.      }  
  33.    }  
  34.   
  35.    // Take each TaskSet in our scheduling order, and then offer it each node in increasing order  
  36.    // of locality levels so that it gets a chance to launch local tasks on all of them.  
  37.    // NOTE: the preferredLocality order: PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY  
  38.    for (taskSet <- sortedTaskSets) {  
  39.      var launchedAnyTask = false  
  40.      var launchedTaskAtCurrentMaxLocality = false  
  41.      for (currentMaxLocality <- taskSet.myLocalityLevels) {  
  42.        do {  
  43.          launchedTaskAtCurrentMaxLocality = resourceOfferSingleTaskSet(  
  44.            taskSet, currentMaxLocality, shuffledOffers, availableCpus, tasks)  
  45.          launchedAnyTask |= launchedTaskAtCurrentMaxLocality  
  46.        } while (launchedTaskAtCurrentMaxLocality)  
  47.      }  
  48.      if (!launchedAnyTask) {  
  49.        taskSet.abortIfCompletelyBlacklisted(hostToExecutors)  
  50.      }  
  51.    }  
  52.   
  53.    if (tasks.size > 0) {  
  54.      hasLaunchedTask = true  
  55.    }  
  56.    return tasks  
  57.  }  

  1. 随机化了有效的executor的列表，为了均匀的分配
  2. 获取池里（前面已经提过油两种池）的排号序的taskSetManager的队列
  3. 对taskSetManager里面的task集合进行调度分配

4.2.1 taskSetManager队列的排序

这里的排序是对单个Pool里的taskSetManager进行排序，Spark有两种排序算法

[java] view plain copy
 print?
  1. var taskSetSchedulingAlgorithm: SchedulingAlgorithm = {  
  2.   schedulingMode match {  
  3.     case SchedulingMode.FAIR =>  
  4.       new FairSchedulingAlgorithm()  
  5.     case SchedulingMode.FIFO =>  
  6.       new FIFOSchedulingAlgorithm()  
  7.     case _ =>  
  8.       val msg = "Unsupported scheduling mode: $schedulingMode. Use FAIR or FIFO instead."  
  9.       throw new IllegalArgumentException(msg)  
  10.   }  
  11. }  

在这里就简单介绍FIFOSchedulingAlgorithm的算法

[java] view plain copy
 print?
  1. private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {  
  2.   override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {  
  3.     val priority1 = s1.priority  
  4.     val priority2 = s2.priority  
  5.     var res = math.signum(priority1 - priority2)  
  6.     if (res == 0) {  
  7.       val stageId1 = s1.stageId  
  8.       val stageId2 = s2.stageId  
  9.       res = math.signum(stageId1 - stageId2)  
  10.     }  
  11.     res < 0  
  12.   }  
  13. }  
这里的priority 就是前面说到的JobID, 也就是JobID越小的排序在前面，在相通JobId下的StageId越小的排序在前面

4.2.2 单个TaskSetManager的task调度

TaskSetManager 里保存了TaskSet，也就是DAGScheduler里生成的tasks的集合，在TaskSchedulerImpl.scala中进行了单个的TaskSetManager进行调度

[java] view plain copy
 print?
  1. private def resourceOfferSingleTaskSet(  
  2.       taskSet: TaskSetManager,  
  3.       maxLocality: TaskLocality,  
  4.       shuffledOffers: Seq[WorkerOffer],  
  5.       availableCpus: Array[Int],  
  6.       tasks: IndexedSeq[ArrayBuffer[TaskDescription]]) : Boolean = {  
  7.     var launchedTask = false  
  8.     for (i <- 0 until shuffledOffers.size) {  
  9.       val execId = shuffledOffers(i).executorId  
  10.       val host = shuffledOffers(i).host  
  11.       if (availableCpus(i) >= CPUS_PER_TASK) {  
  12.         try {  
  13.           for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {  
  14.             tasks(i) += task  
  15.             val tid = task.taskId  
  16.             taskIdToTaskSetManager(tid) = taskSet  
  17.             taskIdToExecutorId(tid) = execId  
  18.             executorIdToRunningTaskIds(execId).add(tid)  
  19.             availableCpus(i) -= CPUS_PER_TASK  
  20.             assert(availableCpus(i) >= 0)  
  21.             launchedTask = true  
  22.           }  
  23.         } catch {  
  24.           case e: TaskNotSerializableException =>  
  25.             logError(s"Resource offer failed, task set ${taskSet.name} was not serializable")  
  26.             // Do not offer resources for this task, but don't throw an error to allow other  
  27.             // task sets to be submitted.  
  28.             return launchedTask  
  29.         }  
  30.       }  
  31.     }  
  32.     return launchedTask  
  33.   }  

在这里，我们看到了一个参数CPUS_PER_TASK

[java] view plain copy
 print?
  1. val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)  
在spark里，我们可以设置task所使用的cpu的数量，默认是1个，一个task任务在executor中是启动一个线程来执行的

通过计算每个executor的剩余资源，决定是否需要从tasksetmanager里分配出task.

[java] view plain copy
 print?
  1. def resourceOffer(  
  2.     execId: String,  
  3.     host: String,  
  4.     maxLocality: TaskLocality.TaskLocality)  
  5.   : Option[TaskDescription] =  
  6. {  
  7.     .....  
  8.   
  9.     dequeueTask(execId, host, allowedLocality).map { case ((index, taskLocality, speculative)) =>  
  10.       ......  
  11.       new TaskDescription(taskId = taskId, attemptNumber = attemptNum, execId,  
  12.         taskName, index, serializedTask)  
  13.     }  
  14.   } else {  
  15.     None  
  16.   }  
  17. }  

核心函数dequeueTask

[java] view plain copy
 print?
  1. private def dequeueTask(execId: String, host: String, maxLocality: TaskLocality.Value)  
  2.   : Option[(Int, TaskLocality.Value, Boolean)] =  
  3. {  
  4.   for (index <- dequeueTaskFromList(execId, host, getPendingTasksForExecutor(execId))) {  
  5.     return Some((index, TaskLocality.PROCESS_LOCAL, false))  
  6.   }  
  7.   
  8.   if (TaskLocality.isAllowed(maxLocality, TaskLocality.NODE_LOCAL)) {  
  9.     for (index <- dequeueTaskFromList(execId, host, getPendingTasksForHost(host))) {  
  10.       return Some((index, TaskLocality.NODE_LOCAL, false))  
  11.     }  
  12.   }  
  13.   
  14.   if (TaskLocality.isAllowed(maxLocality, TaskLocality.NO_PREF)) {  
  15.     // Look for noPref tasks after NODE_LOCAL for minimize cross-rack traffic  
  16.     for (index <- dequeueTaskFromList(execId, host, pendingTasksWithNoPrefs)) {  
  17.       return Some((index, TaskLocality.PROCESS_LOCAL, false))  
  18.     }  
  19.   }  
  20.   
  21.   if (TaskLocality.isAllowed(maxLocality, TaskLocality.RACK_LOCAL)) {  
  22.     for {  
  23.       rack <- sched.getRackForHost(host)  
  24.       index <- dequeueTaskFromList(execId, host, getPendingTasksForRack(rack))  
  25.     } {  
  26.       return Some((index, TaskLocality.RACK_LOCAL, false))  
  27.     }  
  28.   }  
  29.   
  30.   if (TaskLocality.isAllowed(maxLocality, TaskLocality.ANY)) {  
  31.     for (index <- dequeueTaskFromList(execId, host, allPendingTasks)) {  
  32.       return Some((index, TaskLocality.ANY, false))  
  33.     }  
  34.   }  
  35.   
  36.   // find a speculative task if all others tasks have been scheduled  
  37.   dequeueSpeculativeTask(execId, host, maxLocality).map {  
  38.     case (taskIndex, allowedLocality) => (taskIndex, allowedLocality, true)}  
  39. }  

在Spark中为了尽量分配任务到task所需的资源的本地,依据task里的preferredLocations所保存的需要资源的位置进行分配

  1. 尽量分配到task到task所需资源相同的executor里执行，比如ExecutorCacheTaskLocation，HDFSCacheTaskLocation
  2. 尽量分配到task里task所需资源相通的host里执行
  3. task的数组从最后向前开始分配

分配完生成TaskDescription，里面记录着taskId, execId, task在数组的位置，和task的整个序列化的内容


4.2.3 Launch Tasks

[java] view plain copy
 print?
  1. private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {  
  2.       for (task <- tasks.flatten) {  
  3.         val serializedTask = ser.serialize(task)  
  4.         if (serializedTask.limit >= maxRpcMessageSize) {  
  5.           scheduler.taskIdToTaskSetManager.get(task.taskId).foreach { taskSetMgr =>  
  6.             try {  
  7.               var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +  
  8.                 "spark.rpc.message.maxSize (%d bytes). Consider increasing " +  
  9.                 "spark.rpc.message.maxSize or using broadcast variables for large values."  
  10.               msg = msg.format(task.taskId, task.index, serializedTask.limit, maxRpcMessageSize)  
  11.               taskSetMgr.abort(msg)  
  12.             } catch {  
  13.               case e: Exception => logError("Exception in error callback", e)  
  14.             }  
  15.           }  
  16.         }  
  17.         else {  
  18.           val executorData = executorDataMap(task.executorId)  
  19.           executorData.freeCores -= scheduler.CPUS_PER_TASK  
  20.   
  21.           logDebug(s"Launching task ${task.taskId} on executor id: ${task.executorId} hostname: " +  
  22.             s"${executorData.executorHost}.")  
  23.   
  24.           executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))  
  25.         }  
  26.       }  
  27.     }  

这里的逻辑就相对比较简单，TaskDescription里面包含着executorId，而CoarseGrainedSchedulerBackend里有executor的信息，根据executorId获取到executor的通讯端口，发送LunchTask的信息。

这里有个RPC的消息的大小控制，如果序列化的task的内容超过了最大RPC的消息，这个任务会被丢弃

[java] view plain copy
 print?
  1. /** Returns the configured max message size for messages in bytes. */  
  2.   def maxMessageSizeBytes(conf: SparkConf): Int = {  
  3.     val maxSizeInMB = conf.getInt("spark.rpc.message.maxSize", 128)  
  4.     if (maxSizeInMB > MAX_MESSAGE_SIZE_IN_MB) {  
  5.       throw new IllegalArgumentException(  
  6.         s"spark.rpc.message.maxSize should not be greater than $MAX_MESSAGE_SIZE_IN_MB MB")  
  7.     }  
  8.     maxSizeInMB * 1024 * 1024  
  9.   }  


可以看到最大的消息大小是128M，可以通过spark.rpc.message.maxSize进行配置

相关推荐