How Spark Runs on a Cluster

Chapter 2 named the parts: the driver, the executors and the cluster manager. Chapter 3 showed jobs, stages and tasks in the UI.

This chapter puts them in one order. It follows what happens between the moment you call an action and the moment you get the result.

It also answers a practical question: how many tasks can run at the same time?

Spark work has four levels, each one inside the one above:

The DAG scheduler, a part of the driver, cuts each job into stages. It cuts wherever rows must move between tasks, which is a shuffle (chapter 5).

The steps between two shuffles join into one stage. So count the shuffles to guess the stages, not the steps. filter and select add no stage, but groupBy does.

Take the revenue job: filter, then groupBy. It has one shuffle, so two stages. In chapter 3, show() even started 2 jobs, because AQE ran the first stage as its own job.

The task scheduler, another part of the driver, sends each task to an executor. One executor core runs one task at a time. A core is often called a task slot.

In the chapter 2 model, 2 executors have 2 cores each, so 4 slots. The revenue job has 3 partitions, so 3 tasks. Three slots work, and one stays idle.

When a stage has more tasks than slots, the tasks run in waves. A wave is one round in which every slot runs one task:

spark = SparkSession.builder.master("local[2]").getOrCreate()

six = orders.repartition(6)
print(spark.sparkContext.defaultParallelism, six.rdd.getNumPartitions())
print(six.count())

On a real cluster the arithmetic is the same. 10 executors with 4 cores have 40 slots. A stage with 200 tasks runs in 200 / 40 = 5 waves.

The number of partitions decides how busy the cluster is:

Here is the whole path, in order:

1. Your transformations only describe the work. Nothing runs yet.
2. You call an action, such as show() or count().
3. Catalyst, the optimizer, turns your steps into a better plan (chapter 13).
4. The DAG scheduler cuts the plan into stages at each shuffle.
5. The task scheduler sends one task per partition to the free slots.
6. The executors run the tasks, and write and read shuffle data between stages.
7. The last stage ends, and the result goes to the driver.

If one task fails, Spark runs only that task again, on any free slot. It rebuilds the input from lineage or from shuffle files (chapter 4). The rest of the job keeps its work.

On a cluster, Spark tries a task up to 4 times by default (spark.task.maxFailures). After the last failure, the job fails.

Check yourself: a stage has 12 tasks and the cluster has 4 slots. How many waves run? (Answer: 3.)