[WIP][SPARK-3795] Heuristics for dynamically scaling executors

This is part of a bigger effort to provide elastic scaling of executors within a Spark application ([SPARK-3174](https://issues.apache.org/jira/browse/SPARK-3174)). This PR does not provide any functionality by itself; it is a skeleton that is missing a mechanism to be added later in [SPARK-3822](https://issues.apache.org/jira/browse/SPARK-3822).

The design doc can be found at [SPARK-3174](https://issues.apache.org/jira/browse/SPARK-3174) (under "Heuristics for Scaling Executors". This deviates from the design in two ways: (1) the remove policy is significantly simplified by cutting out the exponential increase logic that mirrors the add policy, and (2) the configuration of the add policy is split up into `addExecutorThreshold` and `addExecutorInterval`. The main reason for this is that the user may want to configure how quickly executors are added, but want to keep the trigger condition the same.

Comments and feedback are most welcome, but please keep in mind that this is still WIP! For those of you reviewing this in detail, I highly recommend doing it through your favorite IDE instead of through the diff here.
I'd go for it, especially after reading more of the code for this, since
you sometimes use millis and sometimes use seconds

On Wed, Oct 15, 2014 at 1:59 PM, andrewor14 <notifications@github.com>
wrote:

> In
> core/src/main/scala/org/apache/spark/scheduler/ExecutorAllocationManager.scala:
>
> > + * request to add or remove executors. The mechanism to actually do this will be added separately,
> > + * e.g. in SPARK-3822 for Yarn.
> > + */
> > +private[scheduler] class ExecutorAllocationManager(scheduler: TaskSchedulerImpl) extends Logging {
> > +  private val conf = scheduler.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!")
> > +  }
> > +
> > +  // How frequently to add and remove executors (seconds)
> > +  private val addThreshold =
> > +    conf.getLong("spark.dynamicAllocation.addExecutorThreshold", 60)
>
> Yeah this is a broader issue I have filed at
> https://issues.apache.org/jira/browse/SPARK-3859. I thought about adding
> Seconds to the end but it makes the name really long. I'm still on the
> fence about that.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/2746/files#r18922277>.
>