Spark Core - Execution Engine: Theoretical Quiz

This assessment focuses on Spark engine internals: JVM task scheduling, DAG boundaries, and physical execution stages.

Scenario 1: Jobs, Stages, and Tasks Mapping

The Scenario

A developer submits a PySpark job containing the following RDD execution pipeline:

# Stage 1 Ingestion
rdd1 = sc.textFile("hdfs://cluster/logs/*.log")
rdd2 = rdd1.filter(lambda line: "ERROR" in line).map(lambda line: (line.split()[3], 1))

# Stage 2 Aggregation
rdd3 = rdd2.reduceByKey(lambda a, b: a + b)

# Action 1
rdd3.saveAsTextFile("hdfs://cluster/output/errors")

# Action 2
error_count = rdd3.count()

The Questions

1. How many Jobs are triggered during the execution of this script?
2. Explain the physical distinction between a Stage Boundary and a Task Unit, and identify the stage boundaries in this code.

Detailed Solution & Architectural Analysis

1. Number of Jobs Triggered

Each Action triggered on an RDD or DataFrame launches a separate, dedicated Job through the SparkContext. In this script, there are exactly 2 Actions:

1. .saveAsTextFile(...) Triggers Job 0.
2. .count() Triggers Job 1. Thus, Spark will compile and execute exactly 2 distinct Jobs.

2. Stage Boundaries vs. Task Units

• Stage Boundary: Spark divides Jobs into physical execution blocks called Stages. Stage boundaries are established whenever a Wide Dependency (which forces a shuffle, e.g. reduceByKey) occurs in the execution graph.
  • Stage 1 (Narrow stage): Reads from HDFS, applies .filter(), splits lines, and maps to (key, 1) tuples in-memory. The mappers write shuffle partition files to their local executor disks.
  • Stage 2 (Wide stage): Spark schedules shuffle fetches to read intermediate files from Stage 1 mappers, groups keys, executes the merge lambda function, and writes the output files back to HDFS.
• Task Unit: A Task is the smallest executable unit of computation, representing a single thread executing the Stage instructions on a single physical partition block. If a Stage processes 100 partitions, Spark compiles and schedules 100 identical Tasks to be executed in parallel by worker nodes.

Scenario 2: Shuffle Write & Read Mechanics

The Scenario

During a heavy aggregation stage, a Spark execution log reports: INFO ShuffleBlockFetcherIterator: Spilling 4.2 GB of Shuffle Data to Disk The execution slows down drastically during the wide dependency.

The Questions

1. Explain the differences between Shuffle Write and Shuffle Read operations.
2. Where physically are the intermediate shuffle files stored, and how does this affect recovery if an executor node crashes?

Detailed Solution & Architectural Analysis

1. Shuffle Write vs. Shuffle Read

• Shuffle Write: Occurs at the end of the parent stage (Map stage). Executors write intermediate results locally, grouping rows by target partition numbers. It writes two files: an index file (detailing partition byte offsets) and a data file (containing all serialized records).
• Shuffle Read: Occurs at the start of the child stage (Reduce stage). The target executors query the Driver for block metadata and fetch their assigned partition segments over the network from the map executors' local disks.

2. Shuffle File Storage & Recovery

• Physical Location: Intermediate shuffle files are written to the local scratch disks (configured in spark.local.dir or YARN node manager local paths) of the mapper executors.
• Crash Recovery: If an executor node crashes during Shuffle Read, the data partitions residing on its local scratch disk are permanently lost.
• DAG Re-computation: The Driver detects the missing shuffle blocks and re-runs the entire parent Map stage for those partitions on a healthy node to reconstruct the lost shuffle files. This highlights why shuffles increase recovery costs.

Scenario 3: Task Scheduling and JVM Thread Reuse

The Scenario

A pipeline runs with a cluster setting --executor-cores 4. The YARN execution log shows tasks executing concurrently inside the same executor container process.

The Questions

1. How does the TaskScheduler allocate tasks to executors?
2. What are the advantages of JVM thread reuse inside a single executor over launching separate JVM processes for each task?

Detailed Solution & Architectural Analysis

1. TaskScheduler Allocation

The TaskScheduler receives task sets from the DAGScheduler and coordinates execution:

1. Locality Check: It queries the BlockManager to find which nodes hold the target partitions (preferring PROCESS_LOCAL, then NODE_LOCAL, then RACK_LOCAL).
2. Task Launch: It issues YARN execution requests to launch tasks on matching executors.
3. Executor Slots: An executor with --executor-cores 4 is configured with 4 execution slots. The executor JVM launches 4 parallel threads to execute the tasks concurrently.

2. JVM Thread Reuse Advantages

In old MapReduce systems, every task ran inside a separate JVM process, incurring high JVM startup latency (2-3 seconds per task).

• Executor JVM persistence: Spark executors are persistent JVM processes that run throughout the application lifecycle.
• Thread Execution: Tasks are lightweight threads (java.lang.Thread) scheduled on a shared thread pool inside the executor. Thread initialization takes less than a millisecond, eliminating JVM startup latency and allowing tasks to share cached datasets, broadcast variables, and Tungsten off-heap memory pools directly.

Scenario 4: Speculative Execution for Straggler Tasks

The Scenario

A cluster administrator notices that a batch job is stalled on Task #45, which is running 20x slower than the other tasks due to a hardware slowdown on YARN node #8.

The Questions

1. Explain how Speculative Execution operates to resolve straggler tasks.
2. What are the CPU overhead hazards of enabling speculative execution on non-idempotent sinks?

Detailed Solution & Architectural Analysis

1. Speculative Execution Mechanics

When spark.speculation is enabled (true):

1. The Driver monitors task durations.
2. If one task runs significantly slower than the median duration of completed tasks, Spark assumes the host node is degraded.
3. It launches a duplicate backup instance of the same task on a healthy executor node in parallel.
4. Whichever task completes first is kept; the other task instance is killed by the Driver.

2. Execution Hazards on Non-Idempotent Sinks

• The Hazard: Speculative execution assumes tasks are idempotent (running them twice has no side effects).
• If the task writes to a database or a non-transactional file system without transaction isolation, both task instances will write duplicate rows concurrently, resulting in double inserts and corrupted data.
• Tuning Guideline: Disable speculative execution if you are writing to raw database tables or custom non-atomic output sinks.