Reading Parallelism

Where a stage's task count comes from

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 next animation reads 9 page hits as 3 partitions, and its first line sets the regrouped side to 4 partitions. Count the tasks in each stage, and watch what each task of the second stage receives.

spark.conf.set("spark.sql.shuffle.partitions", 4)
(page_hits.groupBy("device")
   .agg(F.count("*").alias("hits"))
   .collect())

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.

Tasks against slots

A stage's task count says how much work is queued. Your slots, 1 per executor core, say how much runs at once. With 10 executors of 4 cores, 40 tasks run together and the rest wait for a free slot. Each batch is a wave, and when tasks take similar time, a stage lasts about its wave count times 1 task's duration.

The bottom 2 rows are the ones to worry about. At 50 tasks, the second wave runs 10 tasks while 30 slots idle, so the stage takes 2 task-durations for 1.25 waves of work. At 8 tasks, 32 of the 40 cores do nothing for the whole stage, and adding machines can't help until the task count rises.

A cluster has 8 executors with 5 cores each. A stage has 100 tasks that each take roughly 30 seconds. Roughly how long does the stage take?

• About 90 seconds, since 40 slots run the 100 tasks in 3 waves
• About 75 seconds, since 100 tasks over 40 slots is 2.5 waves
• About 30 seconds, since every task gets a core of its own at once
• About 6 minutes, since each of 8 executors works through 12 or 13 tasks

Reading it off the Stages tab

3 readings to take on any slow stage

• Total tasks under the slot count. The stage leaves cores idle by construction. It needs more partitions, and more machines would sit idle too.
• Exactly 200 tasks after a boundary. Nobody set spark.sql.shuffle.partitions, and a count that suits 2GB of regrouped data fits 2TB badly.
• Thousands of tasks lasting tens of milliseconds. Launching and tracking each task costs the driver time, and here that cost rivals the work itself.

Put the readings to work on a single row of the tab: a stage after a groupBy shows 200/200 tasks on a cluster with 40 slots, and its median task took 4 minutes. The count of exactly 200 says the setting was never touched. That is 5 waves of 4 minutes, which puts the stage near 20 minutes. Raising the count to 800 would cut each task to around 1 minute, but 800 tasks is 20 waves, still about 20 minutes, because the work was already spread over all 40 slots. With evenly split data, only more slots shorten this stage; more partitions change how much each task holds in memory.

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.

About this lesson

This section is part of the Execution Model: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.