Dynamic Resource Allocation

Overview

Learn how to configure Spark to allocate and deallocate executors dynamically based on task backlog metrics.

What You Will Learn

In this lesson, you will learn:

Dynamic Scaling: Why scaling executor counts up and down saves money.
Backpressure Configurations: Tuning scaling rules.
External Shuffle Service: Keeping shuffle data safe when workers scale down.

Detailed Concept Explanation

By default, a Spark application retains its allocated executors for the entire duration of its run, even if it is idle.

Dynamic Resource Allocation allows Spark to automatically scale executor counts up and down based on workload needs.

Key Configuration Settings

To enable dynamic allocation, configure these settings:

spark.dynamicAllocation.enabled = true
spark.dynamicAllocation.minExecutors: The minimum number of executors to keep alive.
spark.dynamicAllocation.maxExecutors: The maximum number of executors to scale up to.
spark.dynamicAllocation.schedulerBacklogTimeout: How long task queues must be backed up (default 1s) before Spark spins up new executors.
spark.dynamicAllocation.executorIdleTimeout: How long an executor can be idle (default 60s) before Spark terminates it to free up resources.

The External Shuffle Service

When an executor terminates, any shuffle files stored on its local disk are lost. To prevent this, you must run the External Shuffle Service on each node. This service runs independently of executor processes, allowing worker nodes to serve shuffle files even after their executors have scaled down.

Code Examples

Python (PySpark) Configuration

python

from pyspark.sql import SparkSession

# Build optimal dynamic allocation config settings
spark = SparkSession.builder \
    .appName("DynamicAllocationTest") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "1") \
    .config("spark.dynamicAllocation.maxExecutors", "20") \
    .config("spark.shuffle.service.enabled", "true") \
    .getOrCreate()

df = spark.range(1, 100)
df.collect()

Execution Plan Diagram (Python & Scala)

Execution Plan Diagram

SparkSession.builder

config(dynamicAllocation.enabled

true)

config(minExecutors

config(maxExecutors

20)

getOrCreate()

Scala Configuration

scala

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("DynamicAllocationScala")
  .config("spark.dynamicAllocation.enabled", "true")
  .config("spark.dynamicAllocation.minExecutors", "1")
  .config("spark.dynamicAllocation.maxExecutors", "20")
  .config("spark.shuffle.service.enabled", "true")
  .getOrCreate()

val df = spark.range(1, 100)
df.collect()