Execution Model: Intermediate

Before any data moves, Spark cuts your program into 3 nested units. A job is the work 1 action asks for. A stage is a stretch of that job in which no row has to leave the task holding it. A task is 1 stage applied to 1 partition, and it is the only unit an executor ever runs.

The first row says "at least" because a few operations, orderBy among them, submit an extra job ahead of yours. To pick its value ranges, the sort samples the rows arriving at it, so that sampling job runs every stage upstream of the sort to produce those rows. Spark keeps the shuffle files that job writes, and your job then lists the stages already done as skipped instead of running them again. The stage counts in this section describe the plan of the job your action submits, and the Lazy Evaluation lessons trace the extra job in full.

That plan has 3 stages and 6 tasks for 1 action, and the driver ends up with 210 for US and 145 for EU, with APAC at 75. The Jobs tab shows 2 jobs, though. The sampling job runs stage 0, then 2 tasks that aggregate the regions and sample the result. The main job finds stage 0's shuffle files already written, lists that stage as skipped, and runs stages 1 and 2, so the per-region sums are computed twice and 8 tasks run in all. filter and withColumn never get a stage of their own. Each output row depends on 1 input row, so Spark runs both inside the task that reads the partition and builds the partial sums in the same pass over its rows. A stage ends only where a task would need rows another task holds.

Scale the input to 50GB and the shape stays the same while the numbers grow. Spark reads files in pieces of up to 128MB by default, so the read stage gets around 400 tasks, and it runs once, inside the sampling job. Every stage after a boundary, the sampling pass included, gets 200 tasks, the default value of spark.sql.shuffle.partitions.

These counts assume adaptive execution is off. Spark 3.2 and later turn it on by default and can merge small partitions mid-job, a behaviour with lessons of its own.

Run this on the practice data. It has no groupBy anywhere, only a filter and a new column before the count, yet the Stages tab lists 2 stages for it. Spark counts in 2 halves: each task counts the rows in its own partition, then those per-partition numbers move to a single partition, where 1 task adds them up. That move is a boundary like any other, so the second stage has exactly 1 task.

items = spark.table("order_items")
heavy = (items
    .filter(F.col("quantity") >= 2)
    .withColumn("line_total", F.col("quantity") * F.col("unit_price")))
heavy.count()

Predict before you edit: change the last line to heavy.groupBy("order_id").count().count() and run it. The job grows to 3 stages: the read, the per-order totals, and the single task that counts orders. The number returned changes from line items to orders holding at least 1 such line.

Everything else you have used so far, filter, select, withColumn, union, runs inside the stage already in progress.

This program on the practice data counts line items per order and sorts the orders by that count. Trace it before running: a groupBy boundary, then an orderBy boundary, so 3 stages. The rows come back in descending order across the whole result because each task in the last stage sorted 1 contiguous range of counts, and the driver lines the ranges up.

items = spark.table("order_items")
per_order = (items
    .filter(F.col("order_id").isNotNull())
    .groupBy("order_id")
    .agg(F.count("*").alias("lines"))
    .orderBy(F.col("lines").desc(), F.col("order_id")))
per_order.collect()

Now add .limit(5) to the end of the chain and run it again. The 5 rows match the top of the full result, yet that version runs 2 stages, for the top-N reason in the tip above.

Every figure in the Tasks column of the Stages tab has a source you can name. A stage that reads files gets its partitions from the input: Spark packs file bytes into partitions no larger than spark.sql.files.maxPartitionBytes, which defaults to 128MB. Several small files can end up in 1 partition. A stage that starts after a boundary gets whatever count the boundary split the data into.

The first stage runs 3 tasks because the input has 3 partitions. The second runs 4 because the setting says 4, even though only 3 devices exist, so at least 1 of its tasks receives no rows: it gets scheduled and finishes without finding anything. The empty task leaves the result alone. Mobile comes back with 4 hits and desktop with 3, and tv has the remaining 2. Spark launches a task for every partition whether it holds data, so a groupBy over 12 distinct keys at the default of 200 leaves at least 188 tasks with no rows to process.

If a stage after a boundary shows fewer than 200 tasks, the runtime merging noted earlier is the likely reason: in Spark 3.2 and later, small partitions are combined after Spark measures them.

In client mode, the default, the driver runs inside the spark-submit process on the machine you launched from: your laptop, a shared gateway box, a notebook server. In cluster mode, spark-submit asks the cluster manager to start the driver on one of the cluster's own nodes, and the driver runs there from then on while spark-submit only reports on it.

Client mode's cost is distance. The driver sends a message to launch every task and receives 1 back when it finishes, so a 400-task stage puts at least 800 messages on the link between your machine and the cluster. A collect() that returns 2GB crosses the same link: over a 50 megabit-per-second VPN that is more than 5 minutes of transfer, where a driver inside the cluster's network would take seconds.

A third arrangement runs on a single machine: --master local[4] puts the driver and 4 task threads in 1 process, so your slots are 4 threads, and shuffle pieces are written and read on the same disk with no network between them.

A running application can tell you which arrangement it got. The property spark.submit.deployMode holds client or cluster, and spark.master holds the master URL.

Each named flag of spark-submit is a shortcut for a Spark property, one of the spark.* keys that the Environment tab of the Spark UI lists for a running application. A property with no shortcut goes through --conf key=value, which you can repeat.

One property can be set in 3 places, and Spark settles them in a fixed order. A value set in code on the session builder beats a spark-submit flag or --conf, and a flag beats the cluster's spark-defaults.conf file. Here is 1 job's settings traced through all 3.

Precedence decides which value Spark reads; the other question is when Spark reads it. Spark reads executor memory and cores, along with the executor count, when executors launch, and every executor in an application launches with the values the application started with. Most properties under spark.sql., including spark.sql.shuffle.partitions, are read each time a query is planned, so a new value applies from the next action on.

The session builder fails more quietly. In a notebook where a session already exists, SparkSession.builder.config("spark.executor.memory", "8g").getOrCreate() hands back that existing session, and its executors keep the memory they started with.

Driver memory follows the same rule, with a twist from the previous section. In client mode the driver is the very process running your code, so Spark's documentation says not to set spark.driver.memory in code there: by then the driver's JVM has already started. Pass --driver-memory or put it in spark-defaults.conf. In client mode that heap also comes from the machine you launched on, so a laptop with 16GB of RAM can't host a 24g driver.

When a job behaves as if a setting never landed, open the Environment tab: it lists every property the application started with and the value that won. A value changed later with spark.conf.set lives in the session and doesn't appear there, so check that one through its effect, such as the task count of the first stage after the next boundary.