ExecutorAllocationManager — Allocation Manager for Spark Core

ExecutorAllocationManager is responsible for dynamically allocating and removing executors based on the workload.

It intercepts Spark events using the internal ExecutorAllocationListener that keeps track of the workload (changing the internal registries that the allocation manager uses for executors management).

It uses ExecutorAllocationClient, LiveListenerBus, and SparkConf (that are all passed in when ExecutorAllocationManager is created).

ExecutorAllocationManager is created when SparkContext is created and dynamic allocation of executors is enabled.

Note
 SparkContext expects that SchedulerBackend follows the ExecutorAllocationClient contract when dynamic allocation of executors is enabled.
Table 1. FIXME’s Internal Properties
Name Initial Value Description

executorAllocationManagerSource

ExecutorAllocationManagerSource

FIXME
Table 2. ExecutorAllocationManager’s Internal Registries and Counters
Name Description

executorsPendingToRemove

Internal cache with…​FIXME

Used when…​FIXME

removeTimes

Internal cache with…​FIXME

Used when…​FIXME

executorIds

Internal cache with…​FIXME

Used when…​FIXME

initialNumExecutors

FIXME

numExecutorsTarget

FIXME

numExecutorsToAdd

FIXME

initializing

Flag whether…​FIXME

Starts enabled (i.e. true).
Tip

Enable INFO logging level for org.apache.spark.ExecutorAllocationManager logger to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.ExecutorAllocationManager=INFO

Refer to Logging.

addExecutors Method

Caution
 FIXME

removeExecutor Method

Caution
 FIXME

maxNumExecutorsNeeded Method

Caution
 FIXME

Starting ExecutorAllocationManager — start Method

start(): Unit

start registers ExecutorAllocationListener (with LiveListenerBus) to monitor scheduler events and make decisions when to add and remove executors. It then immediately starts spark-dynamic-executor-allocation allocation executor that is responsible for the scheduling every 100 milliseconds.

Note
 100 milliseconds for the period between successive scheduling is fixed, i.e. not configurable.

It requests executors using the input ExecutorAllocationClient. It requests spark.dynamicAllocation.initialExecutors.

Note
 start is called while SparkContext is being created (with dynamic allocation enabled).

Scheduling Executors — schedule Method

schedule(): Unit

schedule calls updateAndSyncNumExecutorsTarget to…​FIXME

It then go over removeTimes to remove expired executors, i.e. executors for which expiration time has elapsed.

updateAndSyncNumExecutorsTarget Method

updateAndSyncNumExecutorsTarget(now: Long): Int

updateAndSyncNumExecutorsTarget…​FIXME

If ExecutorAllocationManager is initializing it returns 0.

Resetting ExecutorAllocationManager — reset Method

reset(): Unit

reset resets ExecutorAllocationManager to its initial state, i.e.

  1. initializing is enabled (i.e. true).

  2. The currently-desired number of executors is set to the initial value.

  3. The numExecutorsToAdd is set to 1.

  4. All executor pending to remove are cleared.

  5. All ??? are cleared.

Stopping ExecutorAllocationManager — stop Method

stop(): Unit

stop shuts down spark-dynamic-executor-allocation allocation executor.

Note
 stop waits 10 seconds for the termination to be complete.

Creating ExecutorAllocationManager Instance

ExecutorAllocationManager takes the following when created:

ExecutorAllocationManager initializes the internal registries and counters.

Validating Configuration of Dynamic Allocation — validateSettings Internal Method

validateSettings(): Unit

validateSettings makes sure that the settings for dynamic allocation are correct.

validateSettings validates the following and throws a SparkException if not set correctly.

  1. spark.dynamicAllocation.minExecutors must be positive

  2. spark.dynamicAllocation.maxExecutors must be 0 or greater

  3. spark.dynamicAllocation.minExecutors must be less than or equal to spark.dynamicAllocation.maxExecutors

  4. spark.dynamicAllocation.executorIdleTimeout must be greater than 0

  5. spark.shuffle.service.enabled must be enabled.

  6. The number of tasks per core, i.e. spark.executor.cores divided by spark.task.cpus, is not zero.

Note
 validateSettings is used when ExecutorAllocationManager is created.

spark-dynamic-executor-allocation Allocation Executor

spark-dynamic-executor-allocation allocation executor is a…​FIXME

It is started…​

It is stopped…​

results matching ""

    No results matching ""