Who Does What on a Cluster

The Daily Grind has four hundred shops, and head office wants the busiest hour across all of them. One laptop could not hold a year of every till, which is why the Why Spark lesson sent you here. The answer is to share the work out across many machines, and Spark's design is the list of who holds what while that happens. Every word in this lesson is one Spark's own documentation uses, because these are the words you will meet in a log, a stack trace or the Spark UI on your first real job.

The parts

Driver program. The process running your program: the one that builds the SparkSession, holds every DataFrame you define as a description, and turns an action into work for other machines. There is one. Spark's docs call it "the process running the main() function of the application."

Cluster manager. The service the driver asks for machines. It is not Spark; it is Kubernetes, YARN, or Spark's own standalone manager, and its whole job is to hand out resources: "an external service for acquiring resources on the cluster."

Worker node. A machine in the cluster that can run application code. Many.

Executor. A process on a worker node, launched for your application, that runs tasks and keeps data in memory across them. Several per machine, many per application, and they belong to your application alone: "each application has its own executors."

At the top, two boxes: the driver program, which holds your program and the plan and sends tasks, and the cluster manager, which hands out machines. Below them three worker nodes, each running an executor; inside each executor are its tasks, one per data partition, highlighted: two tasks on worker node 1, two on worker node 2, one on worker node 3. Arrows run from the driver to the cluster manager, labelled asks for executors, and from the driver to the executors, labelled sends code and tasks.

driver programyour main(), the SparkSession, the plan · one

cluster managerhands out machines · Kubernetes, YARN, standalone

↓ asks for executors↓ sends code, then tasks

worker node 1a machine

executortask · partition 1task · partition 2

worker node 2a machine

executortask · partition 3task · partition 4

worker node 3a machine

executortask · partition 5

The driver asks the cluster manager for executors, sends them your code, and then sends them tasks. The results of those tasks flow back as the answer to whatever action you called.

How the work is cut up

Task. A unit of work sent to one executor. A task processes one data partition: the chunk of rows the Where Your Code Runs lesson, next, is about.

Job. Everything Spark does in response to one action. count(), collect(), show(), a write: each one is a job. Transformations never start one; that is the Nothing Runs Until You Ask rule seen from the cluster's side.

Stage. A job is cut into stages at every point where rows have to change machines, the shuffle. Within a stage, every task can run without waiting for any other; between stages, every task of the next has to wait for the shuffle to finish. The docs: "each job gets divided into smaller sets of tasks called stages that depend on each other."

A job box for the action count() on orders filtered to large then grouped by size. Inside it, stage 1: read, filter and pre-count each partition, with three tasks, one per partition. Then a highlighted shuffle boundary labelled rows move so each size is in one place. Then stage 2: finish the counts, with two tasks, one for large and one for small and medium.

job · one action: .count()orders.filter(size == "large").groupBy("size").count()

stage 1read, filter and pre-count each partition, all at once

task 1 · partition 1task 2 · partition 2task 3 · partition 3

shuffle — rows move so each size is in one place; stages are cut here

stage 2finish the counts, one task per group of keys

task 1 · size = largetask 2 · size = small, medium

So orders.filter(...).groupBy("size").count() is one job with two stages: the first reads and filters and pre-totals every partition in parallel, then the shuffle moves rows so that each size is in one place, then the second stage finishes the totals. The number of tasks in a stage is the number of partitions it reads. The number of stages in a job is the number of shuffles plus one.

Where this site fits

Your code here is PySpark talking Spark Connect: the client sends the plan you built, not your Python, to a server over gRPC, and the server sends rows back. In Spark Connect, the driver lives on the server, not in your process. On this site the server is Sail, running in its local mode: one process that plays driver and executor in one, on one machine. The anatomy is the same shape with every box on the same machine, which is why the plans you print here are honest about structure and silent about cost. When the Where Your Code Runs lesson shows a shuffle, you now know which boxes the rows would be moving between.

The symptom, and the fix

The anatomy explains the first crash most people meet. collect() is an action whose job ends by sending every row to the driver, the one process; on a result bigger than the driver's memory the job dies with an out-of-memory error on the driver, while four hundred executors sit idle with plenty of room. The fix is to keep rows where the executors are: write the result to storage, keep collect() for small answers, and use limit or show when you only want to look. The driver is for coordination; it was never meant to hold the data.

Self-check

orders.filter(F.col("size") == "large").groupBy("drink").agg(F.count("*")).collect() runs on a table of 200 partitions. How many jobs, and how many stages?

One job: collect() is the only action. Two stages: there is one shuffle, for the groupBy, and stages are shuffles plus one. The first stage runs 200 tasks, one per partition, each filtering and pre-counting its own rows; the second finishes the counts after the rows for each drink have been brought together.

Nothing to grade here. Next, Where Your Code Runs prints a plan and shows you the shuffle those stages are cut at.