4

Spark 中每个工作节点会启动多少个执行器？我能知道它背后的数学原理吗？

例如，我有 6 个工作节点和 1 个主节点，如果我通过 spark-submit 提交作业，将为作业启动多少个执行器？

4

2 回答 2

4

捎带@LiMuBei的回答......

首先，它是你告诉它的任何东西

--num-executors 4

如果使用动态分配，那么这就是它为您决定的方式

根据这份文件（http://jerryshao.me/architecture/2015/08/22/spark-dynamic-allocation-investigation/），

Spark 如何通过挂起和正在运行的任务计算它需要的最大执行器数量：

 private def maxNumExecutorsNeeded(): Int = {
    val numRunningOrPendingTasks = listener.totalPendingTasks + listener.totalRunningTasks
    (numRunningOrPendingTasks + tasksPerExecutor - 1) / tasksPerExecutor
 }

如果当前执行者数量大于预期数量：

 // The target number exceeds the number we actually need, so stop adding new
 // executors and inform the cluster manager to cancel the extra pending requests
 val oldNumExecutorsTarget = numExecutorsTarget
 numExecutorsTarget = math.max(maxNeeded, minNumExecutors)
 numExecutorsToAdd = 1

 // If the new target has not changed, avoid sending a message to the cluster manager
 if (numExecutorsTarget < oldNumExecutorsTarget) {
   client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks, hostToLocalTaskCount)
   logDebug(s"Lowering target number of executors to $numExecutorsTarget (previously " +
     s"$oldNumExecutorsTarget) because not all requested executors are actually needed")
 }
 numExecutorsTarget - oldNumExecutorsTarget

如果当前 executor 数量大于期望的数量，Spark 将通知集群管理器取消挂起的请求，因为它们是不需要的。对于那些已经分配的执行者，稍后将通过超时机制将它们降低到合理的数量。

如果当前执行者数量不能满足所需数量：

 val oldNumExecutorsTarget = numExecutorsTarget

 // There's no point in wasting time ramping up to the number of executors we already have, so
 // make sure our target is at least as much as our current allocation:
 numExecutorsTarget = math.max(numExecutorsTarget, executorIds.size)

 // Boost our target with the number to add for this round:
 numExecutorsTarget += numExecutorsToAdd

 // Ensure that our target doesn't exceed what we need at the present moment:
 numExecutorsTarget = math.min(numExecutorsTarget, maxNumExecutorsNeeded)

 // Ensure that our target fits within configured bounds:
 numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)
 val delta = numExecutorsTarget - oldNumExecutorsTarget

 // If our target has not changed, do not send a message
 // to the cluster manager and reset our exponential growth
 if (delta == 0) {
   numExecutorsToAdd = 1
   return 0
 }
 val addRequestAcknowledged = testing ||
   client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks, hostToLocalTaskCount)
 if (addRequestAcknowledged) {
   val executorsString = "executor" + { if (delta > 1) "s" else "" }
   logInfo(s"Requesting $delta new $executorsString because tasks are backlogged" +
     s" (new desired total will be $numExecutorsTarget)")
   numExecutorsToAdd = if (delta == numExecutorsToAdd) {
     numExecutorsToAdd * 2
   } else {
     1
   }
   delta
 } else {
   logWarning(
     s"Unable to reach the cluster manager to request $numExecutorsTarget total executors!")
   0
 }
于 2016-10-11T01:46:23.690 回答
2

两个可能的答案：

  • 如果您在调用时指定执行者的数量，spark-submit您应该得到您要求的数量--num-executors X
  • 如果您未指定，则默认情况下 Spark 应使用动态分配，如果需要，它将启动更多执行程序。在这种情况下，您可以配置行为，例如执行者的最大数量，请参阅http://spark.apache.org/docs/latest/configuration.html#dynamic-allocation

每个工作节点的执行器数量将取决于可用的资源。

于 2016-10-10T22:07:49.547 回答