Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/2746#discussion_r19452708
--- Diff:
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala ---
@@ -0,0 +1,413 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark
+
+import scala.collection.mutable
+
+import org.apache.spark.scheduler._
+
+/**
+ * An agent that dynamically allocates and removes executors based on the
workload.
+ *
+ * The add policy depends on whether there are backlogged tasks waiting to
be scheduled. If
+ * the scheduler queue is not drained in N seconds, then new executors are
added. If the queue
+ * persists for another M seconds, then more executors are added and so
on. The number added
+ * in each round increases exponentially from the previous round until an
upper bound on the
+ * number of executors has been reached.
+ *
+ * The rationale for the exponential increase is twofold: (1) Executors
should be added slowly
+ * in the beginning in case the number of extra executors needed turns out
to be small. Otherwise,
+ * we may add more executors than we need just to remove them later. (2)
Executors should be added
+ * quickly over time in case the maximum number of executors is very high.
Otherwise, it will take
+ * a long time to ramp up under heavy workloads.
+ *
+ * The remove policy is simpler: If an executor has been idle for K
seconds, meaning it has not
+ * been scheduled to run any tasks, then it is removed.
+ *
+ * There is no retry logic in either case because we make the assumption
that the cluster manager
+ * will eventually fulfill all requests it receives asynchronously.
+ *
+ * The relevant Spark properties include the following:
+ *
+ * spark.dynamicAllocation.enabled - Whether this feature is enabled
+ * spark.dynamicAllocation.minExecutors - Lower bound on the number of
executors
+ * spark.dynamicAllocation.maxExecutors - Upper bound on the number of
executors
+ *
+ * spark.dynamicAllocation.schedulerBacklogTimeout (M) -
+ * If there are backlogged tasks for this duration, add new executors
+ *
+ * spark.dynamicAllocation.sustainedSchedulerBacklogTimeout (N) -
+ * If the backlog is sustained for this duration, add more executors
+ * This is used only after the initial backlog timeout is exceeded
+ *
+ * spark.dynamicAllocation.executorIdleTimeout (K) -
+ * If an executor has been idle for this duration, remove it
+ */
+private[spark] class ExecutorAllocationManager(sc: SparkContext) extends
Logging {
+ import ExecutorAllocationManager._
+
+ private val conf = sc.conf
+
+ // Lower and upper bounds on the number of executors. These are required.
+ private val minNumExecutors =
conf.getInt("spark.dynamicAllocation.minExecutors", -1)
+ private val maxNumExecutors =
conf.getInt("spark.dynamicAllocation.maxExecutors", -1)
+ if (minNumExecutors < 0 || maxNumExecutors < 0) {
+ throw new SparkException("spark.dynamicAllocation.{min/max}Executors
must be set!")
+ }
+ if (minNumExecutors > maxNumExecutors) {
+ throw new SparkException("spark.dynamicAllocation.minExecutors must " +
+ "be less than or equal to spark.dynamicAllocation.maxExecutors!")
+ }
+
+ // How long there must be backlogged tasks for before an addition is
triggered
+ private val schedulerBacklogTimeout = conf.getLong(
+ "spark.dynamicAllocation.schedulerBacklogTimeout", 60)
+
+ // Same as above, but used only after `schedulerBacklogTimeout` is
exceeded
+ private val sustainedSchedulerBacklogTimeout = conf.getLong(
+ "spark.dynamicAllocation.sustainedSchedulerBacklogTimeout",
schedulerBacklogTimeout)
+
+ // How long an executor must be idle for before it is removed
+ private val removeThresholdSeconds = conf.getLong(
+ "spark.dynamicAllocation.executorIdleTimeout", 600)
+
+ // Number of executors to add in the next round
+ private var numExecutorsToAdd = 1
+
+ // Number of executors that have been requested but have not registered
yet
+ private var numExecutorsPending = 0
+
+ // Executors that have been requested to be removed but have not been
killed yet
+ private val executorsPendingToRemove = new mutable.HashSet[String]
+
+ // All known executors
+ private val executorIds = new mutable.HashSet[String]
+
+ // A timestamp of when an addition should be triggered, or NOT_SET if it
is not set
+ // This is set when pending tasks are added but not scheduled yet
+ private var addTime: Long = NOT_SET
+
+ // A timestamp for each executor of when the executor should be removed,
indexed by the ID
+ // This is set when an executor is no longer running a task, or when it
first registers
+ private val removeTimes = new mutable.HashMap[String, Long]
+
+ // Polling loop interval (ms)
+ private val intervalMillis: Long = 100
+
+ /**
+ * Register for scheduler callbacks to decide when to add and remove
executors.
+ */
+ def start(): Unit = {
+ val listener = new ExecutorAllocationListener(this)
+ sc.addSparkListener(listener)
+ startPolling()
+ }
+
+ /**
+ * Start the main polling thread that keeps track of when to add and
remove executors.
+ * During each loop interval, this thread checks if the time then has
exceeded any of the
+ * add and remove times that are set. If so, it triggers the
corresponding action.
+ */
+ private def startPolling(): Unit = {
+ val t = new Thread {
+ override def run(): Unit = {
+ while (true) {
+ ExecutorAllocationManager.this.synchronized {
+ val now = System.currentTimeMillis
+ try {
+ // If the add time has expired, add executors and refresh
the add time
+ if (addTime != NOT_SET && now >= addTime) {
+ addExecutors()
+ logDebug(s"Starting timer to add more executors (to " +
+ s"expire in $sustainedSchedulerBacklogTimeout seconds)")
+ addTime += sustainedSchedulerBacklogTimeout * 1000
+ }
+
+ // If any remove time has expired, remove the corresponding
executor
+ removeTimes.foreach { case (executorId, expireTime) =>
+ if (now > expireTime) {
+ removeExecutor(executorId)
+ removeTimes.remove(executorId)
+ }
+ }
+ } catch {
+ case e: Exception => logError("Exception in dynamic executor
allocation thread!", e)
+ }
+ }
+ Thread.sleep(intervalMillis)
+ }
+ }
+ }
+ t.setName("spark-dynamic-executor-allocation")
+ t.setDaemon(true)
+ t.start()
+ }
+
+ /**
+ * Request a number of executors from the cluster manager.
+ * If the cap on the number of executors is reached, give up and reset
the
+ * number of executors to add next round instead of continuing to double
it.
+ * Return the number actually requested. Exposed for testing.
+ */
+ def addExecutors(): Int = synchronized {
--- End diff --
Rather than expose all these methods for testing, can you use
`PrivateMethodTester`?
http://doc.scalatest.org/1.7/org/scalatest/PrivateMethodTester.html
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
