Dynamic Resource Allocation

In Apache Spark, dynamic resource allocation lets a running application adjust how much compute it holds: it returns executors it is no longer using and asks for more when demand rises. Several applications can use the returned capacity when they share a cluster.

How it behaves

When tasks remain pending beyond spark.dynamicAllocation.schedulerBacklogTimeout, Spark requests more executors. Persistent backlog triggers further requests after spark.dynamicAllocation.sustainedSchedulerBacklogTimeout. The target stays within minExecutors and maxExecutors; requested capacity still needs to be supplied by the cluster manager. For example, with a maximum of ten executors, a growing queue cannot make this application acquire an eleventh.

• It is off by default. Besides spark.dynamicAllocation.enabled, the deployment must provide a way to keep shuffle output usable when an executor goes away. Shuffle tracking, which is on by default, satisfies this. If tracking is turned off, one of the other arrangements must be configured instead: an external shuffle service, decommissioning, or a shuffle plugin backed by reliable storage. Decommissioning needs both spark.decommission.enabled and spark.storage.decommission.shuffleBlocks.enabled.
• Executors are removed after an idle period, governed by spark.dynamicAllocation.executorIdleTimeout, which defaults to 60 seconds.
• Executors holding cached data are not removed by default, to avoid losing that data. A separate timeout controls whether they ever are.

Why shuffle data is the complication

During a shuffle, each executor writes map output to local disk for other executors to fetch. Removing the executor can make that output unavailable before downstream tasks finish using it. Lost output must then be recomputed, so allocation needs a way to preserve access or delay removal.

The mechanisms differ in whether they retain the executor, serve files from its node, or move the data before removal. The three below cover those alternatives; a custom shuffle plugin backed by reliable storage provides another option.

• An external shuffle service serves files after the executor is gone. A separate process on the worker node serves the shuffle files and outlives individual executors. The files stay on the original node, so losing that node can still lose the output.
• Shuffle tracking keeps the executor alive. Spark retains executors that hold shuffle data needed by active jobs. The tracking timeout defaults to infinity, so removal normally waits for the shuffles to be garbage collected. This can keep an apparently idle executor longer than its ordinary idle timeout.
• Decommissioning migrates blocks before removal. Graceful decommissioning tries to move cached RDD and shuffle blocks to another executor. The original executor can then leave while surviving executors hold the data. Migration is an attempt rather than a guarantee; blocks lost when shutdown occurs before migration finishes may require recomputation.

The external service and successful migration can retain access after executor removal, at different locations. Tracking instead delays removal while the data remains needed.

It matters when reading a cluster that is not shrinking as expected. With an external service, executors going away while shuffle files remain readable is the intended behavior. With tracking, executors that look idle but refuse to be released are also the intended behavior, and the lever is the tracking timeout rather than the idle timeout. Diagnosing the second as a bug in dynamic allocation is the common mistake.

With the default tracking left on, this requirement is already met. If tracking is turned off and no other arrangement is configured, the configuration is invalid rather than merely slow: the allocation manager checks the settings at startup and refuses to start. The recomputation cost appears in a valid configuration when shuffle files are actually lost — a migration that did not finish before shutdown, or a node that died — and then the cluster spends money re-deriving data it already had.

Where it pays, and where it does not

It is also worth separating this from cloud autoscaling, which adds and removes machines. Dynamic allocation moves executors between applications within whatever cluster exists; if the underlying nodes are fixed and reserved, returning executors reduces contention but not the bill. The two are usually worth configuring together, and each has its own delay before capacity arrives.

How this fits with skew, memory, and the rest of what decides a job’s time and cost is worked through in Why More Executors Stopped Helping.

References

• Apache Spark documentation, Job Scheduling
• Apache Spark Documentation, Configuration