Spark Scheduling - Strategy

Driver

Standalone(Master)

/**
 * Schedule the currently available resources among waiting apps. This method will be called
 * every time a new app joins or resource availability changes.
 */
private def schedule(): Unit = {
  if (state != RecoveryState.ALIVE) {
    return
  }
  // Drivers take strict precedence over executors
  val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0
  for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
    // We assign workers to each waiting driver in a round-robin fashion. For each driver, we
    // start from the last worker that was assigned a driver, and continue onwards until we have
    // explored all alive workers.
    var launched = false
    var numWorkersVisited = 0
    while (numWorkersVisited < numWorkersAlive && !launched) {
      val worker = shuffledAliveWorkers(curPos)
      numWorkersVisited += 1
      if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
        launchDriver(worker, driver)
        waitingDrivers -= driver
        launched = true
      }
      curPos = (curPos + 1) % numWorkersAlive
    }
  }
  startExecutorsOnWorkers()
}

Yarn(AM)

@VisibleForTesting
public static final class ScheduleTransition
    implements
    MultipleArcTransition<RMAppAttemptImpl, RMAppAttemptEvent, RMAppAttemptState> {
  @Override
  public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
      RMAppAttemptEvent event) {
    ApplicationSubmissionContext subCtx = appAttempt.submissionContext;
    if (!subCtx.getUnmanagedAM()) {
      // Need reset #containers before create new attempt, because this request
      // will be passed to scheduler, and scheduler will deduct the number after
      // AM container allocated
      
      // Currently, following fields are all hard coded,
      // TODO: change these fields when we want to support
      // priority or multiple containers AM container allocation.
      for (ResourceRequest amReq : appAttempt.amReqs) {
        amReq.setNumContainers(1);
        amReq.setPriority(AM_CONTAINER_PRIORITY);
      }

      int numNodes =
          RMServerUtils.getApplicableNodeCountForAM(appAttempt.rmContext,
              appAttempt.conf, appAttempt.amReqs);
      if (LOG.isDebugEnabled()) {
        LOG.debug("Setting node count for blacklist to " + numNodes);
      }
      appAttempt.getAMBlacklistManager().refreshNodeHostCount(numNodes);

      ResourceBlacklistRequest amBlacklist =
          appAttempt.getAMBlacklistManager().getBlacklistUpdates();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Using blacklist for AM: additions(" +
            amBlacklist.getBlacklistAdditions() + ") and removals(" +
            amBlacklist.getBlacklistRemovals() + ")");
      }
      // AM resource has been checked when submission
      Allocation amContainerAllocation =
          appAttempt.scheduler.allocate(
              appAttempt.applicationAttemptId,
              appAttempt.amReqs,
              EMPTY_CONTAINER_RELEASE_LIST,
              amBlacklist.getBlacklistAdditions(),
              amBlacklist.getBlacklistRemovals(),
              new ContainerUpdates());
      if (amContainerAllocation != null
          && amContainerAllocation.getContainers() != null) {
        assert (amContainerAllocation.getContainers().size() == 0);
      }
      return RMAppAttemptState.SCHEDULED;
    } else {
      // save state and then go to LAUNCHED state
      appAttempt.storeAttempt();
      return RMAppAttemptState.LAUNCHED_UNMANAGED_SAVING;
    }
  }
}

Executor

Standalone(Master)

/**
 * Schedule executors to be launched on the workers.
 * Returns an array containing number of cores assigned to each worker.
 *
 * There are two modes of launching executors. The first attempts to spread out an application's
 * executors on as many workers as possible, while the second does the opposite (i.e. launch them
 * on as few workers as possible). The former is usually better for data locality purposes and is
 * the default.
 *
 * The number of cores assigned to each executor is configurable. When this is explicitly set,
 * multiple executors from the same application may be launched on the same worker if the worker
 * has enough cores and memory. Otherwise, each executor grabs all the cores available on the
 * worker by default, in which case only one executor per application may be launched on each
 * worker during one single schedule iteration.
 * Note that when `spark.executor.cores` is not set, we may still launch multiple executors from
 * the same application on the same worker. Consider appA and appB both have one executor running
 * on worker1, and appA.coresLeft > 0, then appB is finished and release all its cores on worker1,
 * thus for the next schedule iteration, appA launches a new executor that grabs all the free
 * cores on worker1, therefore we get multiple executors from appA running on worker1.
 *
 * It is important to allocate coresPerExecutor on each worker at a time (instead of 1 core
 * at a time). Consider the following example: cluster has 4 workers with 16 cores each.
 * User requests 3 executors (spark.cores.max = 48, spark.executor.cores = 16). If 1 core is
 * allocated at a time, 12 cores from each worker would be assigned to each executor.
 * Since 12 < 16, no executors would launch [SPARK-8881].
 */
private def scheduleExecutorsOnWorkers(
    app: ApplicationInfo,
    usableWorkers: Array[WorkerInfo],
    spreadOutApps: Boolean): Array[Int] = {
  val coresPerExecutor = app.desc.coresPerExecutor
  val minCoresPerExecutor = coresPerExecutor.getOrElse(1)
  val oneExecutorPerWorker = coresPerExecutor.isEmpty
  val memoryPerExecutor = app.desc.memoryPerExecutorMB
  val numUsable = usableWorkers.length
  val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker
  val assignedExecutors = new Array[Int](numUsable) // Number of new executors on each worker
  var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)

  /** Return whether the specified worker can launch an executor for this app. */
  def canLaunchExecutor(pos: Int): Boolean = {
    val keepScheduling = coresToAssign >= minCoresPerExecutor
    val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor

    // If we allow multiple executors per worker, then we can always launch new executors.
    // Otherwise, if there is already an executor on this worker, just give it more cores.
    val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutors(pos) == 0
    if (launchingNewExecutor) {
      val assignedMemory = assignedExecutors(pos) * memoryPerExecutor
      val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
      val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
      keepScheduling && enoughCores && enoughMemory && underLimit
    } else {
      // We're adding cores to an existing executor, so no need
      // to check memory and executor limits
      keepScheduling && enoughCores
    }
  }

  // Keep launching executors until no more workers can accommodate any
  // more executors, or if we have reached this application's limits
  var freeWorkers = (0 until numUsable).filter(canLaunchExecutor)
  while (freeWorkers.nonEmpty) {
    freeWorkers.foreach { pos =>
      var keepScheduling = true
      while (keepScheduling && canLaunchExecutor(pos)) {
        coresToAssign -= minCoresPerExecutor
        assignedCores(pos) += minCoresPerExecutor

        // If we are launching one executor per worker, then every iteration assigns 1 core
        // to the executor. Otherwise, every iteration assigns cores to a new executor.
        if (oneExecutorPerWorker) {
          assignedExecutors(pos) = 1
        } else {
          assignedExecutors(pos) += 1
        }

        // Spreading out an application means spreading out its executors across as
        // many workers as possible. If we are not spreading out, then we should keep
        // scheduling executors on this worker until we use all of its resources.
        // Otherwise, just move on to the next worker.
        if (spreadOutApps) {
          keepScheduling = false
        }
      }
    }
    freeWorkers = freeWorkers.filter(canLaunchExecutor)
  }
  assignedCores
}

Job

def initialize(backend: SchedulerBackend) {
  this.backend = backend
  schedulableBuilder = {
    schedulingMode match {
      case SchedulingMode.FIFO =>
        new FIFOSchedulableBuilder(rootPool)
      case SchedulingMode.FAIR =>
        new FairSchedulableBuilder(rootPool, conf)
      case _ =>
        throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
        s"$schedulingMode")
    }
  }
  schedulableBuilder.buildPools()
}

• FIFO

• FAIR

Task

• Local

• Lazy

Links

• Spark Documentation

---

• Author：HyperJ
• Source：HyperJ’s Blog
• Link：Spark Scheduling - Strategy