第17课：Spark Streaming资源动态申请和动态控制消费速率原理剖析

// Optionally scale number of executors dynamically based on workload. Exposed for testing.val dynamicAllocationEnabled = Utils.isDynamicAllocationEnabled(_conf)if (!dynamicAllocationEnabled && //参数配置是否开启资源动态分配_conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
  logWarning("Dynamic Allocation and num executors both set, thus dynamic allocation disabled.")
}

_executorAllocationManager =
  if (dynamicAllocationEnabled) {    Some(new ExecutorAllocationManager(this, listenerBus, _conf))
  } else {    None
  }
_executorAllocationManager.foreach(_.start())

/** * This is called at a fixed interval to regulate the number of pending executor requests * and number of executors running. * * First, adjust our requested executors based on the add time and our current needs. * Then, if the remove time for an existing executor has expired, kill the executor. * * This is factored out into its own method for testing. */private def schedule(): Unit = synchronized {  val now = clock.getTimeMillis

  updateAndSyncNumExecutorsTarget(now)

  removeTimes.retain { case (executorId, expireTime) =>
    val expired = now >= expireTime    if (expired) {
      initializing = false
      removeExecutor(executorId)
    }    !expired
  }
}

/** * Register for scheduler callbacks to decide when to add and remove executors, and start * the scheduling task. */def start(): Unit = {
  listenerBus.addListener(listener)  val scheduleTask = new Runnable() {    override def run(): Unit = {      try {
        schedule()
      } catch {        case ct: ControlThrowable =>
          throw ct        case t: Throwable =>
          logWarning(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)
      }
    }
  }// intervalMillis定时器触发时间
  executor.scheduleAtFixedRate(scheduleTask, 0, intervalMillis, TimeUnit.MILLISECONDS)
}