Performance Tuning & Debugging with the Spark UI

A senior interview question: "This job takes four hours. Make it one." First you find where the time goes.

The Spark UI shows that. It is a web page that Spark serves while your job runs.

In the earlier chapters you met skew, spill, shuffles and partitions. This chapter shows how to tell which one hurts your job.

The chapter has six parts.

1. The order to read the Spark UI.
2. How to read the task table.
3. How to read the plan with its numbers.
4. Seven signs in the UI, and the fix for each.
5. A full diagnosis (finding the cause) of a four-hour job, and the fix.
6. The History Server, and a tuning checklist.

Do not change a setting first. Look first. Tasks is not a tab: you open it from a stage.

Go through the Spark UI in these seven steps:

1. Jobs: which job took the most time? A job is work that an action starts, such as a write.
2. Stages: which stage took the most time? A stage is the part of a job between two shuffles.
3. Tasks: open the stage. Read its shuffle read, shuffle write, spill, Summary Metrics and event timeline.
4. SQL / DataFrame: open the query. It shows the plan with numbers on each step.
5. Executors: check GC time, memory and failed tasks. GC means garbage collection: Java frees unused memory, and tasks pause while it runs.
6. Storage: see which data is cached, and how much of it fits in memory.
7. Environment: see the settings that were set. A missing setting uses its default.

The first four steps narrow the search from a job to one step of the plan. The last three check the machines, the cache and the settings.

You stop as soon as you can name the cause. Most jobs have one stage that takes most of the time. Start there.

A task is the work on one partition. The chapter "Data Skew & Salting" showed the Summary Metrics table. It lists Min, 25th percentile, Median, 75th percentile and Max for the tasks in the stage.

It counts finished tasks only. While a slow task still runs, sort the task list by Duration.

The shape of the Duration row tells you the problem. Here are four shapes. The numbers are made up.

Look at the other rows too. Shuffle Read Size / Records shows how much data each task read. Spill is data that did not fit in memory.

Spill (Memory) is the size in memory. Spill (Disk) is the smaller, packed size on disk.

The event timeline draws one bar for each task over time. One very long bar means one slow task. A big empty space on the right means cores had no work while they waited.

The chapter "Spark SQL & the Catalyst Optimizer" showed plans without numbers. The SQL / DataFrame tab draws the same plan, and puts the real numbers on each step.

We use the shop. The query adds up paid orders for each country. These are the orders and the customers.

(orders.filter(F.col("status") == "paid")
       .join(customers, "customer_id")
       .groupBy("country")
       .agg(F.sum("amount").alias("total"))
       .show())

An Exchange is a shuffle: rows move between machines. A broadcast means Spark copies the small table to every executor.

The UI draws the plan as boxes. We write it as simplified text. Read it from the bottom up, because the data flows upward.

Here is the plan for this query, with the numbers.

HashAggregate (final)                   number of output rows: 3
+- Exchange hashpartitioning(country)
   +- HashAggregate (partial)
      +- BroadcastHashJoin [customer_id]    number of output rows: 4
         :- Filter (status = paid)          number of output rows: 4
         :  +- Scan orders                  number of output rows: 6
         +- BroadcastExchange (small: 5 rows)
            +- Scan customers               number of output rows: 5

Two numbers matter most when the data is large. The first is the rows out of a join. If a join gives far more rows than went in, a key repeats on both sides.

The second is the Exchange. Its data size and shuffle records show how much data crosses the network. If one side of a join is small, a broadcast can replace the shuffle, as in this plan.

On its own, Spark broadcasts a table only below a size limit, 10 MB by default. If customers grows past it, the join type changes. You see this in the worked example.

Some fixes use AQE (Adaptive Query Execution: Spark changes the plan while it runs).

Each problem leaves a sign in the UI. The table gives seven signs, each with its cause and the first fix to try.

Two signs need more words. First, a stage retry.

A fetch failure means a task could not read shuffle files from another machine. Spark reruns the lost parts of the earlier stage, then retries this stage.

This costs time, so find out why the machine was lost. A common cause is an executor that used more memory than it was given.

Second, the driver. The call collect() brings every row to the driver, which is one machine. While the driver works, no stage runs, so you see a gap between jobs.

The setting spark.driver.maxResultSize stops a result that is too big. Its default is 1 GB. The job fails with an error before the driver runs out of memory.

You met this limit in "Spark Architecture: Driver, Executors & Cluster Managers".

Now one full diagnosis. All numbers here are made up for teaching. They are not measurements.

Imagine the shop has 1 billion paid orders and 2 million customers. To keep the numbers simple, we count only paid orders. A daily job joins them and adds up the total by country.

The job takes 4 hours, which is 240 minutes. Step 1, Jobs tab: Job 5, the write, takes 230 of the 240 minutes.

Step 2, Stages tab: Stage 12 takes 212 minutes and has 200 tasks.

Step 3, open the 200 tasks of Stage 12. The Summary Metrics table shows this.

The longest task takes 210 minutes. The median task takes 45 seconds. That is 280 times longer.

The longest task also read 100 million rows, which is one tenth of all 1 billion paid rows. Sort the task list by Spill: it is the only task that spills. This is skew, and the spill is a result of it.

Now find the cause and the fix.

1. Step 4, SQL tab: the join is a SortMergeJoin (both sides are shuffled and sorted by the key). Both sides have an Exchange. Customers is 80 MB.
2. Step 5, Executors tab: GC time is small and no tasks failed. Memory is not the main cause.
3. Steps 6 and 7, Storage and Environment tabs: nothing is cached, and spark.sql.adaptive.enabled is false. The broadcast limit is not listed, so it is the default 10 MB.
4. Last, a count by customer_id: one business account, customer 7, has 100 million orders.

The cause is the hot key, customer 7. A sort-merge join sends all of its rows to one task.

The customers table is 80 MB, over the 10 MB broadcast limit. So Spark picked a sort-merge join, as its rules say. AQE checks the real size, so it also would not broadcast 80 MB.

But AQE would split the hot partition. It is on by default since Spark 3.2, so someone turned it off here. You have two fixes: turn AQE back on, or broadcast customers on purpose.

We choose the second fix: broadcast customers on purpose. With AQE, 1 billion orders still cross the network. With a broadcast, they do not move for the join.

The driver and every executor then hold a full copy of customers. In memory, that copy can be a few times bigger than 80 MB, but it still fits.

report = (orders.filter(F.col("status") == "paid")
                .join(F.broadcast(customers), "customer_id")
                .groupBy("country")
                .agg(F.sum("amount").alias("total")))

The hot customer’s 100 million rows now stay in the many partitions where they were read. No task gets all of them.

Run the job again and open the same tabs. The numbers are made up, as before.

Compare the same fields before and after. The long task is gone, the spill is gone, and the join stage is gone. The join now runs inside the scan stage.

The Spark UI lives only while the application (your whole Spark program) runs. When it ends, the page is gone. A run that failed last night leaves nothing to look at.

The History Server fixes this. Spark writes event logs while it runs. The History Server reads them and shows the same Jobs, Stages and SQL pages later.

# On every job:
spark.eventLog.enabled           true
spark.eventLog.dir               s3a://shop-logs/spark-events
# On the History Server:
spark.history.fs.logDirectory    s3a://shop-logs/spark-events

Turn event logs on for every production job. Many cloud Spark services, for example Amazon EMR, already run a History Server for you. Check before you build your own.

Start with the cheap, safe changes. Do the settings last. This is the order.

1. Data first. Filter early, read only the columns you need, and fix many small files.
2. Partitions. Make tasks neither tiny nor too big. Check skew.
3. Joins. Broadcast the small side. Let AQE handle what it can.
4. Memory. Fix spill and high GC time. See "Memory Management & Tungsten".
5. Settings last. Change one setting at a time and measure again.

Settings come last because most settings only move a limit. A bad join or reading too much data stays slow with a better setting.

Four points to keep.

• Look before you change anything. Go from Jobs to Stages to Tasks to the plan, then Executors, Storage and Environment.
• The shape of the task table names the problem: one long task, many tiny tasks, spill or high GC time.
• Change one thing, then read the same fields again to see if it worked.
• Tune data first, then partitions, joins and memory. Settings come last.

Answer these without looking back.

1. In Summary Metrics, the Max duration is 280 times the Median. What do you suspect?
2. In the plan example, why does the row count go from 6 to 4?
3. A job has 20,000 tasks of 30 milliseconds each. What is the likely cause?
4. Why does the History Server need event logs?
5. Why do you change settings last?