Definition
- Dynamic allocation –
spark.dynamicAllocation.enabled=truelets Spark automatically scale executor count up and down during a job based on pending task backlog and idle timeouts. - Scale up – when tasks are queued beyond
schedulerBacklogTimeout, Spark requests more executors, ramping up exponentially. - Scale down – when an executor has been idle beyond
executorIdleTimeout, Spark releases it back to the cluster. - Prerequisite – shuffle files must survive executor removal: typically an external shuffle service, or shuffle tracking in newer Spark.
At the National Bank of India dynamic allocation is like a branch that calls in extra tellers when a queue forms and sends tellers home when it is quiet, instead of paying a fixed full roster all day. But there is a catch: if a teller who is sent home was holding paperwork other tellers still need (shuffle files), sending them home loses that paperwork. So the branch keeps a shared filing desk (external shuffle service) where departing tellers leave their paperwork.
Understanding the Concepts
What is dynamic resource allocation?
Dynamic resource allocation lets a Spark application automatically scale its executor count up and down during its run based on demand, instead of holding a fixed set for the whole job.
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("NBI-Dynamic-Alloc") \
.config("spark.dynamicAllocation.enabled", "true") \
.config("spark.dynamicAllocation.minExecutors", "2") \
.config("spark.dynamicAllocation.maxExecutors", "50") \
.config("spark.dynamicAllocation.initialExecutors", "5") \
.config("spark.dynamicAllocation.executorIdleTimeout", "60s") \
.config("spark.dynamicAllocation.schedulerBacklogTimeout", "1s") \
.config("spark.shuffle.service.enabled", "true") \
.getOrCreate()
print("Dynamic allocation:", spark.conf.get("spark.dynamicAllocation.enabled"))
print("Min executors:", spark.conf.get("spark.dynamicAllocation.minExecutors"))
print("Max executors:", spark.conf.get("spark.dynamicAllocation.maxExecutors"))
Dynamic allocation: true
Min executors: 2
Max executors: 50
How does scaling up work?
When there are pending tasks that cannot be scheduled because all executors are busy, and that backlog persists beyond the scheduler backlog timeout, Spark requests additional executors. It ramps up exponentially to react quickly to sudden load.
# Scaling up logic:
# 1. Tasks are queued, all executor slots are full
# 2. Backlog persists for > schedulerBacklogTimeout (1s)
# 3. Spark requests new executors from cluster manager
# 4. Ramp-up is exponential: 1, 2, 4, 8, 16...
# (up to maxExecutors)
# Example: heavy aggregation suddenly needs many tasks
import pyspark.sql.functions as F
txn = spark.read.parquet("hdfs:///bank/transactions/2026/")
result = txn.groupBy("branch", "account_id").agg(F.sum("amount"))
result.write.parquet("hdfs:///bank/output/")
# During the shuffle stage, Spark may scale from 5 to 50 executors
How does scaling down work?
When an executor has had no tasks for the idle timeout, Spark releases it back to the cluster manager. This frees resources for other applications.
# Scaling down logic:
# 1. Executor has been idle for > executorIdleTimeout (60s)
# 2. Spark releases it back to cluster manager
# 3. Other applications can use the freed resources
# Between stages or after a job completes,
# idle executors are gradually released
Why is the external shuffle service required?
In a normal shuffle, each executor writes shuffle output to its local disk. If dynamic allocation removes an idle executor, those shuffle files vanish and the dependent stage fails. The external shuffle service is a long-lived process on each worker node that serves shuffle files independently of any executor.
# Without external shuffle service:
# Executor 3 writes shuffle files to its local disk
# Executor 3 becomes idle → dynamic allocation removes it
# Stage 2 needs Executor 3's shuffle files → files gone → FAIL
# With external shuffle service:
# Executor 3 writes shuffle files to its local disk
# External shuffle service on the same node can serve those files
# Executor 3 removed → shuffle files still available via the service
spark.conf.set("spark.shuffle.service.enabled", "true")
# On YARN: runs inside the NodeManager
# On K8s/Standalone: deployed as a per-node process
What alternatives exist to the external shuffle service?
Newer Spark offers alternatives for environments where deploying the external service is awkward.
# Alternative 1: Shuffle tracking (Spark 3.0+)
# Keeps executors that still hold needed shuffle files alive
spark.conf.set("spark.dynamicAllocation.shuffleTracking.enabled", "true")
# No external service needed, but executors stay longer
# Alternative 2: Remote/disaggregated shuffle service
# Pushes shuffle data off executors to a separate storage tier
# Fully decouples compute scaling from shuffle data
# Used in cloud-native deployments (e.g., Uber's RSS, LinkedIn's Magnet)
When does dynamic allocation help?
It shines for interactive or bursty workloads where demand varies. In shared clusters, releasing idle executors lets other applications use them.
# Good for:
# - Interactive notebooks (heavy query, then idle, then another query)
# - Ad-hoc analytics (unpredictable workload)
# - Jobs with alternating heavy and light stages
# - Shared clusters (release idle resources for others)
# Not helpful for:
# - Steady, uniformly loaded batch jobs (all executors busy throughout)
# - Short jobs (ramp-up delay may exceed job duration)
# Example: notebook workflow
# Query 1: heavy join → scales to 50 executors
# Idle: reading results → scales down to 2
# Query 2: light filter → stays at 2-5 executors
What are the tuning parameters?
The main parameters control the executor range, timeouts, and initial count.
# Essential parameters:
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "2") # floor
spark.conf.set("spark.dynamicAllocation.maxExecutors", "100") # ceiling
spark.conf.set("spark.dynamicAllocation.initialExecutors", "10") # start
spark.conf.set("spark.dynamicAllocation.executorIdleTimeout", "120s") # when to release
spark.conf.set("spark.dynamicAllocation.schedulerBacklogTimeout", "1s") # when to request
# Tuning tips:
# - Too aggressive idle timeout → executor churn (repeated startup cost)
# - Too lax idle timeout → holding idle resources too long
# - Too low maxExecutors → cannot scale for heavy stages
# - Too high maxExecutors → can starve other applications
What are common mistakes?
The most common mistake is enabling dynamic allocation without the shuffle-preservation prerequisite.
# WRONG: dynamic allocation without shuffle service
spark.conf.set("spark.dynamicAllocation.enabled", "true")
# spark.shuffle.service.enabled is false
# → Executor removed → shuffle files lost → stage fails
# CORRECT: enable both
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.shuffle.service.enabled", "true")
# WRONG: dynamic allocation for a short, steady job
# Ramp-up delay may exceed the job's total runtime
# Fixed allocation is simpler and just as efficient
What should a senior engineer know?
A senior engineer enables dynamic allocation for shared and bursty workloads with the shuffle-preservation prerequisite satisfied, tunes the idle timeout to avoid churn, and leaves it off for dedicated steady jobs.
# Senior decision:
# Shared cluster, interactive/bursty → dynamic allocation ON
# Dedicated cluster, steady ETL → dynamic allocation OFF (fixed)
# Kubernetes without external shuffle → use shuffle tracking
# Cloud-native → consider remote shuffle service
Visual Overview
Dynamic Resource Allocation:
Time ──────────────────────────────────────────────────→
Executors:
50 │ ┌──────┐
│ │ Heavy│
│ ┌─────┤Stage │
│ scale│ up │ │
10 │ ┌────────┘ └──┬───┘
│ │initial │scale
5 │────┘ │down ┌────┐
│ └────────┤Idle │
2 │─── min ─────────────────────────┴────┴── min ──
└────────────────────────────────────────────────→
Shuffle File Preservation:
┌──────────┐ ┌──────────────────────┐
│Executor 3│ │External Shuffle Svc │
│writes │────→│(long-lived, per node)│
│shuffle │ │serves files after │
│files │ │executor is removed │
└──────────┘ └──────────────────────┘
│
▼ removed (idle)
Shuffle files still available via service
Key Takeaway
Dynamic resource allocation lets a Spark application automatically scale its executor count up and down during its run based on demand, instead of holding a fixed set for the whole job. With spark.dynamicAllocation.enabled set to true, Spark requests more executors when there is a persistent backlog of pending tasks, ramping up quickly, and releases executors that have been idle beyond a timeout, bounded by configured minimum and maximum counts.
The central prerequisite is preserving shuffle output when an executor is removed. In a normal shuffle each executor writes shuffle files to its local disk, so if dynamic allocation removed an executor still holding needed files, the dependent stage would fail. The classic solution is an external shuffle service: a long-lived process on each worker node that serves shuffle files independently of any executor. Newer Spark offers alternatives: shuffle tracking keeps executors with needed files alive, and remote shuffle services push data off executors entirely.
At the National Bank of India I enabled dynamic allocation with the external shuffle service for shared ad-hoc analytics, where jobs needed many executors during a heavy aggregation but sat idle between queries, so scaling down returned resources to other teams. For steady nightly ETL that used all its executors throughout, I left it off, because the scaling machinery and ramp-up delay added overhead for no benefit.
The nuance I stress is that dynamic allocation shines for interactive and bursty workloads and requires the shuffle-preservation prerequisite, and for dedicated steady jobs a simple fixed allocation is just as efficient.
Remember one sentence: dynamic allocation scales executors up on task backlog and down on idle timeout, but requires an external shuffle service or shuffle tracking so removing executors does not lose shuffle files.
---