Spark Architecture: Driver, Executors & Cluster Managers

In the chapter "What Apache Spark Is & Why It Exists" you saw why Spark exists. Big data does not fit on one machine, so Spark splits it into partitions and uses a cluster. A cluster is a group of machines that work together.

This chapter shows who does what when a Spark job runs. You will follow the shop's revenue job from your code to the machines and back.

Later chapters cover the rest. The lazy evaluation chapter explains why Spark waits before it runs. The shuffle chapter explains how rows move between machines.

• Driver and executors: who plans, and who works.
• A walk-through of one job with the shop data.
• Why collect() can crash the driver.
• Memory, failed executors, cluster managers, and deploy modes.

A Spark application has two kinds of process. A process is a program that is running.

An application is your whole program while it runs. It has one driver. A job is one piece of work the application asks for, such as this revenue report.

Here are the five words you need.

Think of a kitchen. The driver is the head cook. The head cook reads the recipe and gives each cook a part of the work.

The executors are the cooks who do the work.

The driver should not hold the big data. The executors work on it. This is the main idea of the chapter.

In the chapter "What Apache Spark Is & Why It Exists" you saw the SparkSession. It is the object your code uses to talk to Spark. It lives in the driver.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("shop").getOrCreate()
orders = spark.read.csv("orders.csv", header=True, inferSchema=True)

Your code never talks to an executor directly. You give the steps to the session. The driver turns them into tasks.

In a notebook, such as Databricks, the session is often made for you. The name is usually spark.

Now follow the shop's revenue job. The chapter "What Apache Spark Is & Why It Exists" used 2 partitions as a model. This time we use a model with 3 partitions of 2 orders each.

Our cluster has 2 executors with 2 cores each. That gives 2 × 2 = 4 task slots. A task slot is a core that can run one task.

For the first step, the driver makes 3 tasks, one for each partition. Three tasks fit in 4 slots, so one slot stays empty. Here is one possible plan.

Each task runs the same steps on its own rows: keep paid orders, then add up the amount for each customer. Look at the three partitions and the code.

from pyspark.sql import functions as F

revenue = (
    orders
    .filter(F.col("status") == "paid")
    .groupBy("customer_id")
    .agg(F.sum("amount").alias("revenue"))
)

Spark then combines the partial results. In real data, one customer can have rows in many partitions, so this step is always needed. The shuffle chapter shows how the rows move.

Last, show() sends the final rows to the driver. The driver prints them.

revenue.show()

collect() is a command that brings every row to the driver. The driver keeps them all in its own memory, as a Python list.

For our 3 revenue rows, that is fine. Now make an assumption: the shop has 500 million orders, and you collect the orders, not the revenue.

all_orders = orders.collect()   # every row goes to the driver

The driver is one process with limited memory. Spark also has a safety limit on the size of results sent to the driver. Its setting is spark.driver.maxResultSize, and the default is 1 GB.

Past the limit, Spark stops this job with an error that names the setting. The driver keeps running.

Below the limit, the driver can still run out of memory, because rows get bigger once the driver unpacks them. Then the driver stops.

Use a command that keeps the data on the executors, or that returns only a few rows.

Each executor has a fixed amount of memory. You set it when you start the application. The setting is spark.executor.memory.

All the cores in one executor share that memory. An executor with 2 cores runs 2 tasks at the same time. Both tasks use the same memory.

More cores in one executor means less memory for each task.

Spark splits this memory into two main parts. Execution memory is for work like grouping and sorting. Storage memory keeps saved (cached) data.

An executor also gets a little extra memory, called overhead. It is for things outside Spark itself, like Python. The default is 10% of spark.executor.memory, at least 384 MB.

When execution memory is full, Spark first writes some data to disk. This is called a spill. The executor fails with an out-of-memory error only when it cannot spill.

Machines fail. An executor can run out of memory, or its machine can stop. Spark expects this.

Say executor 1 dies while it runs task 1 on partition 1. The driver notices that executor 1 stopped answering.

The driver does not start over. It sends the lost tasks to another executor. Executor 2 runs task 1 on partition 1 again.

Executor 2 has only 2 cores, and task 2 still uses one. So task 0 waits for a free core.

When a task fails, Spark runs it again. If the same task fails 4 times, Spark stops the job. The setting is spark.task.maxFailures.

The driver asks the cluster manager for executors. The cluster manager is a program that owns the machines. It gives your application cores and memory, and takes them back when your application ends.

Spark works with any of these managers. Three are common.

On Databricks you do not choose one. Databricks manages the cluster for you. You pick a size, and the executors appear.

Older guides also name Mesos. Spark no longer supports it.

spark-submit is the command that starts a Spark program from a terminal.

The deploy mode says where the driver runs. You set it when you submit a job. The executors always run in the cluster.

Here is the same job submitted both ways. --master yarn tells Spark to use YARN. Only the deploy mode flag changes. The file name revenue_job.py is made up for this example.

spark-submit --master yarn --deploy-mode client  revenue_job.py
spark-submit --master yarn --deploy-mode cluster revenue_job.py

The chapter "Your First Spark Job & the Spark UI" shows spark-submit in more detail.