Definition
- Speculative execution – a Spark feature (
spark.speculation=true) that detects stragglers and launches duplicate copies on other executors; the first to finish wins. - Detection – a task is flagged if it runs longer than
spark.speculation.multipliertimes the median afterspark.speculation.quantileof the stage has completed. - Enable for – environmental slowness: failing disks, noisy cloud neighbors, heterogeneous hardware.
- Disable for – data skew: the duplicate processes the same oversized partition and is equally slow, wasting resources.
At the National Bank of India imagine ten clerks each processing an equal stack of slips, but one clerk is stuck because their calculator is faulty. Speculative execution is the supervisor noticing that one clerk is far behind and quietly assigning the same stack to a spare clerk at a working desk, then taking whichever finishes first. This is great when the delay is the faulty equipment. But if the clerk is slow because they were handed a stack ten times bigger (skew), giving that same giant stack to a second clerk does not help.
Understanding the Concepts
What is speculative execution?
Speculative execution is Spark's mechanism for handling stragglers: tasks that run much slower than their peers in the same stage. With spark.speculation=true, Spark launches duplicate copies of straggler tasks on other executors and takes whichever copy finishes first.
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("NBI-Speculation") \
.config("spark.speculation", "true") \
.config("spark.speculation.multiplier", "1.5") \
.config("spark.speculation.quantile", "0.75") \
.config("spark.speculation.interval", "100ms") \
.getOrCreate()
# spark.speculation = true: enable speculation
# spark.speculation.multiplier = 1.5: flag tasks > 1.5x median
# spark.speculation.quantile = 0.75: wait until 75% of tasks finish
# spark.speculation.interval = 100ms: check frequency
How does Spark detect stragglers?
Spark waits until a configurable fraction of tasks in the stage have finished (quantile), then computes the median running time. Any still-running task slower than the median times the multiplier is flagged as a straggler.
# Detection logic (simplified):
# 1. Wait until 75% of tasks in the stage have completed
# 2. Compute median task duration of completed tasks
# 3. For each still-running task:
# if task_duration > median * 1.5:
# launch speculative copy on different executor
# 4. Whichever copy (original or speculative) finishes first: keep it
# 5. Kill the other copy
# Example:
# 100 tasks in a stage, 75 have finished
# Median duration of finished tasks: 30 seconds
# Task 42 has been running for 60 seconds (> 30 * 1.5 = 45)
# → Task 42 is flagged, speculative copy launched
How does Spark ensure no duplicate output?
Because two copies of a task may run simultaneously, Spark uses an output commit coordinator. Only one copy is allowed to commit its results, preventing duplicate or corrupted output.
# Output commit safety:
# - Two copies of task 42 are running
# - Copy A finishes first
# - Commit coordinator allows Copy A to commit
# - Copy B is killed
# - Result: exactly one copy's output is committed
# This is why speculation is safe for correctness:
# No duplicate writes, no corrupted output
When should you enable speculative execution?
Enable it when slowness is environmental and task-independent: heterogeneous clusters, noisy cloud neighbors, failing disks, or shared YARN/Kubernetes environments.
# Good candidates for speculation:
spark = SparkSession.builder \
.appName("NBI-Shared-Cluster-Job") \
.config("spark.speculation", "true") \
.config("spark.speculation.multiplier", "2.0") \
.config("spark.speculation.quantile", "0.9") \
.getOrCreate()
# Use case 1: Heterogeneous cluster (some nodes are older/slower)
# Use case 2: Cloud cluster with noisy neighbors
# Use case 3: Shared YARN cluster with occasional node overload
# Use case 4: Hardware with intermittent disk failures
# Conservative settings for production:
# multiplier=2.0 (only flag tasks > 2x median, avoid false positives)
# quantile=0.9 (wait until 90% done before speculating)
When should you disable speculative execution?
Disable it when slowness is caused by data skew. The speculative duplicate processes the same oversized partition and is equally slow, wasting an executor slot.
# BAD: speculation on a skewed stage
# Task 42 is slow because its partition has 5M rows (others have 25K)
# Speculative copy also gets 5M rows → equally slow
# Net result: wasted executor slot, possibly slower overall
# Correct fix for skew:
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
# Or: salt the hot key, isolate and broadcast
How do you diagnose environmental vs skew stragglers?
Inspect the Stages tab in the Spark UI. Compare the straggler's input size and shuffle read to the median.
# Diagnostic:
# Open Stages tab → click on the slow stage → Task details
#
# ENVIRONMENTAL straggler:
# Task 42: Input=128 MB, Duration=180s
# Median: Input=125 MB, Duration=30s
# → Similar input, much longer duration
# → Node is slow (bad disk, noisy neighbor)
# → Speculation HELPS
#
# SKEW straggler:
# Task 42: Input=2.5 GB, Duration=180s
# Median: Input=125 MB, Duration=30s
# → Input is 20x larger → data skew
# → Speculation does NOT help
# → Fix the skew instead
What are the tuning parameters?
The main parameters control how aggressively speculation triggers.
# spark.speculation = true/false
# spark.speculation.multiplier = 1.5 (default)
# Task must be > multiplier * median to be flagged
# Higher = more conservative (fewer false positives)
#
# spark.speculation.quantile = 0.75 (default)
# Fraction of tasks that must complete before speculation starts
# Higher = waits longer, more data to judge
#
# spark.speculation.interval = 100ms (default)
# How often Spark checks for stragglers
# Conservative production settings:
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "2.0")
spark.conf.set("spark.speculation.quantile", "0.9")
What are common mistakes?
The most common mistake is enabling speculation on skewed workloads, which wastes resources. Another is setting the multiplier too low, causing unnecessary duplicates.
# WRONG: speculation on a skewed join (wastes executor slots)
spark.conf.set("spark.speculation", "true")
# Task 42 has 5M rows, duplicate also gets 5M rows → both slow
# WRONG: multiplier too low (false positives)
spark.conf.set("spark.speculation.multiplier", "1.0")
# Every task slightly slower than median gets a duplicate → resource waste
# CORRECT: conservative settings + fix skew separately
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", "2.0")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
What should a senior engineer know?
A senior engineer treats speculation as tail-latency insurance for flaky infrastructure, not a substitute for fixing skew. They tune it conservatively and distinguish environmental from data stragglers before deciding.
# Senior rule:
# 1. Inspect straggler's input size vs median in Spark UI
# 2. Similar input, longer duration → environmental → speculation helps
# 3. Much larger input → skew → fix the skew, speculation is wrong tool
# 4. Tune multiplier and quantile conservatively
# 5. Never rely on speculation to fix a design problem
Visual Overview
Speculative Execution:
Stage with 10 tasks (normal tasks finish in ~30s):
┌────┐┌────┐┌────┐┌────┐┌────┐┌────┐┌────┐┌────┐┌────┐┌────────┐
│ T0 ││ T1 ││ T2 ││ T3 ││ T4 ││ T5 ││ T6 ││ T7 ││ T8 ││T9 slow │
│30s ││28s ││32s ││29s ││31s ││30s ││27s ││33s ││30s ││ 120s? │
└────┘└────┘└────┘└────┘└────┘└────┘└────┘└────┘└────┘└────────┘
straggler!
After quantile (75%) tasks finish:
median = 30s, threshold = 30 * 1.5 = 45s
T9 running for 60s > 45s → flagged!
Launch speculative copy:
┌────────┐ ┌────────────┐
│T9 orig │ │T9 specul. │ (on different executor)
│(slow │ │(hopefully │
│ node) │ │ fast node) │
└────────┘ └─────┬──────┘
│ finishes first!
▼
Commit coordinator → accept speculative copy, kill original
HELPS: environmental slowness (bad node)
WASTES: data skew (same big partition on both copies)
Key Takeaway
Speculative execution is Spark's mechanism for handling stragglers. With spark.speculation set to true, Spark waits until a configurable fraction of a stage's tasks have finished, computes the median completion time, and flags any still-running task slower than the median by more than a multiplier. It launches a duplicate on a different executor, accepts whichever copy finishes first, and kills the other. An output commit coordinator ensures only one copy commits.
The decision to use it hinges on why a task is slow. When the slowness is environmental: a failing disk, a noisy neighbor, or a heterogeneous cluster, the straggler is bad luck on one node, and relaunching it elsewhere lets it finish on healthy hardware. That is the intended use. But when the slowness is data skew, the speculative duplicate processes the same oversized partition and is equally slow, so speculation wastes an executor slot and can make the job slower.
At the National Bank of India I enabled speculation on nightly jobs that ran on a shared heterogeneous cluster where a node occasionally had a slow disk, and it kept stragglers from holding up the stage. But I was careful to disable or ignore it where the straggler also had a much larger input size, because that signalled skew.
The rule I give in interviews is to inspect the Stages tab and compare the straggler's input size to the median: similar input means environmental slowness where speculation helps, much larger input means skew where speculation is the wrong tool. Speculation is tail-latency insurance for flaky infrastructure, not a substitute for fixing skew.
Remember one sentence: speculative execution relaunches stragglers on other executors for environmental slowness, but wastes resources on data skew because the duplicate processes the same oversized partition.
---