# Find the slow step of a Databricks job: task timelines, the query profile and explain

Measure before you tune. Read how long each task of a job takes over its recent runs from system.lakeflow.job_task_run_timeline, then open the slowest task's queries: on serverless, the query profile shows each operator's time, rows and how much of the data was pruned; locally, the Spark UI shows the same per stage. Then read the plan with explain('formatted') for the usual causes: a scan of every file, a shuffle join where a broadcast would do, a skewed key, many small files, or the same work done twice.

- Page: https://tecyfy.com/data-engineering/course/performance-and-cost/find-the-slow-step
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 12: Performance and cost, Find the slow step
- Parts of this module: Serverless or classic compute (https://tecyfy.com/data-engineering/course/performance-and-cost); Find the slow step (https://tecyfy.com/data-engineering/course/performance-and-cost/find-the-slow-step); Build it: what the job costs a month (https://tecyfy.com/data-engineering/course/performance-and-cost/build-it)

At the end of this part you can:

- Rank a job's tasks by the time they take over their recent runs, from the jobs system tables
- Find the slowest operator of a query with the query profile or the Spark UI
- Read a physical plan for the common causes of a slow step, and pick the fix for each

[Part 1](/data-engineering/course/performance-and-cost) chose the compute. Before changing anything else, find out where a run spends its time. Tuning a step that takes 10 seconds of a 10-minute job saves nothing, however clever the change. This part needs the shop's job deployed by [module 11](/data-engineering/course/testing-and-ci-cd/build-it) for the Databricks queries, and [module 3's bigger shop](/data-engineering/course/how-spark-runs-a-job) for the local plan. To rebuild from nothing, run module 3's `make_shop.py`.

## Rank the tasks

Databricks records every task run in the system table `system.lakeflow.job_task_run_timeline`, with its start and end time and its result. This query ranks the shop job's tasks by the time they took over the last 30 days:

```sql
-- How long each task of shop_daily took, over its runs of the last 30 days.
WITH task_runs AS (
  SELECT job_run_id, task_key,
         sum(timestampdiff(SECOND, period_start_time, period_end_time)) AS seconds,
         sum(setup_duration_seconds) AS setup_seconds
  FROM system.lakeflow.job_task_run_timeline
  WHERE job_id IN (SELECT job_id FROM system.lakeflow.jobs WHERE name = 'shop_daily')
    AND period_start_time >= current_date() - INTERVAL 30 DAYS
  GROUP BY job_run_id, task_key
)
SELECT task_key,
       count(*) AS runs,
       round(avg(seconds)) AS avg_seconds,
       max(seconds) AS max_seconds,
       round(avg(setup_seconds)) AS avg_setup_seconds
FROM task_runs
GROUP BY task_key
ORDER BY avg_seconds DESC
```

2 details make it right. A run longer than an hour is split into 1 row per hour, so the inner query adds up each task run's rows before averaging. And a job id is unique only within a workspace, and `system.lakeflow.jobs` keeps a row for every change to a job, so the job is found by name with `IN`; in an account with several workspaces, add `workspace_id` to both. Your development copy is named `[dev <your name>] shop_daily`, so it is not counted.

Read the result in 3 ways. The task at the top is where to look first. A task whose `max_seconds` is far above its average has bad days, often the days with the most data or a retry. And a large `avg_setup_seconds` is time spent getting compute before any work starts: on serverless standard mode, part 1's few minutes of start-up show up here, and no change to the code will move them.

The system tables are read with SQL in the workspace, by users an admin has given access to the `system` catalog. Records usually appear within an hour of a run.

## Find the slow operator

A task runs 1 or more queries. On serverless, open the task's run, then its queries, and each query's **query profile**: a graph of the operators the query ran, with the time each spent and the rows it produced. The **Top operators** panel lists the slowest. The scan operators also show how much of the table was **pruned**, the share of files never read. On your computer, the **SQL** tab of the Spark UI, from [module 3](/data-engineering/course/how-spark-runs-a-job), shows the same graph with each operator's rows and time.

What you are looking for is the 1 operator that took most of the time, and whether it did more work than the query needed: a scan that read every file for a question about 1 day, or a join that produced far more rows than went into it.

## Read the plan before it runs

`explain` prints the plan Spark will run, without running it. With module 3's bigger shop:

```python
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[4]").appName("plan").getOrCreate()
orders = spark.read.parquet("lake/orders")
customers = spark.read.parquet("lake/customers")
by_country = (
    orders.where("amount > 50")
    .join(customers, "customer_id")
    .groupBy("country_id").agg(F.sum("amount").alias("sales"))
)
by_country.explain("formatted")
```

`formatted` prints the plan as a numbered outline, then the details of each step. The outline:

```text
== Physical Plan ==
AdaptiveSparkPlan (11)
+- HashAggregate (10)
   +- Exchange (9)
      +- HashAggregate (8)
         +- Project (7)
            +- BroadcastHashJoin Inner BuildRight (6)
               :- Filter (2)
               :  +- Scan parquet  (1)
               +- BroadcastExchange (5)
                  +- Filter (4)
                     +- Scan parquet  (3)
```

Read it from the bottom, where the data comes in, to the top, and look up each number's details below it. The names to look for:

- **`Scan parquet`** with **`PushedFilters`**: the filters handed to the reader, here `GreaterThan(amount,50.00)` on step 1, which let it skip row groups. A filter that is missing here, and appears only in a `Filter` above the scan, is applied after every row was read.
- **`Exchange`**: a shuffle. Step 9's details show `hashpartitioning` on `country_id` into 200 partitions (the `#` number after a column is Spark's own id, and yours may differ): the partial sums move across the network to be added up by country. There is no shuffle by `customer_id`, because the join did not need one.
- **`BroadcastHashJoin`** or **`SortMergeJoin`**: the join Spark chose. A broadcast join sends the small side to every task, with no shuffle of the large side; Spark picks it when one side is under 10 MB by default. `customers` is small, so this plan broadcasts it, in steps 5 and 6.
- **`AdaptiveSparkPlan isFinalPlan=false`** at the top: [adaptive query execution](/data-engineering/course/how-spark-runs-a-job/adaptive-query-execution) will look at the real sizes as it runs and may change the plan, for example to a broadcast join or by merging small shuffle partitions.

## The usual causes, and their fixes

Most slow steps come down to a few causes. Each shows in the profile or the plan:

| What you see | The cause | The fix |
| --- | --- | --- |
| A scan reads every file, with little pruned | The files cannot be skipped for the filter | Cluster the table on the filter's column; [module 4](/data-engineering/course/delta-lake-and-iceberg/optimize-and-vacuum) |
| A scan reads many tiny files | Small files from frequent small writes | Compact them with `OPTIMIZE`, or let predictive optimization do it; module 4 |
| A `SortMergeJoin` with a small side | Spark did not know the side was small | Check its statistics, or hint `broadcast`; [module 3](/data-engineering/course/how-spark-runs-a-job/adaptive-query-execution) |
| 1 task far slower than the rest of its stage | [Shuffle and Data Skew](https://tecyfy.com/concepts/data-skew): 1 key holds most rows | Adaptive query execution splits skewed sort-merge joins; otherwise split the key |
| A join returns more rows than went in | A key that is not unique on 1 side | Fix the key; [module 8's fan-out](/data-engineering/course/serving-data) |
| The same input read and computed twice | 2 actions on 1 DataFrame | Write the result to a table once and read it; caching is not available on serverless |

Change 1 thing, run the job, and read the same query again. The number to compare is the one you started from: the task's time in the timeline, and the operator's time in the profile.

The [Spark shuffle partition calculator](/data-engineering/spark-shuffle-partitions) shows how a shuffle's partitions, adaptive query execution and a skewed key decide how long a stage waits on its slowest task.

[Part 3](/data-engineering/course/performance-and-cost/build-it) turns the time a run takes into what it costs, from the shop job's own runs.

## Check yourself

1. The query profile of the shop's slowest query shows a scan of orders_daily that read every file of the table, for a query about 1 day. What is the fix to try first?

   A. Add more compute
   B. Lay the table out so the files can be skipped by order_date, then check that the query filters on that column
   C. Cache the table
   D. Raise spark.sql.shuffle.partitions

### Answers

1. B. A scan that reads every file for 1 day means no file could be skipped, because each file holds every day. Clustering the table by order_date, as module 4 showed, lets the reader skip files by their minimum and maximum dates. More compute reads the same files faster at a higher price, and caching is not available on serverless.

## Questions

### How do I find which task of a Databricks job is slow?

Query system.lakeflow.job_task_run_timeline for the job's id over its recent runs, add up each task run's time, and rank the tasks by their average. The table has 1 row per task run, or per hour of a run longer than an hour, with its start and end time, result and set-up time. Then open the slowest task's run and its query profile to find the slow operator.

### What does explain do in PySpark?

DataFrame.explain prints the plan Spark will run for a DataFrame without running it. The default prints the physical plan; explain('formatted') prints an outline of the physical plan and then each operator's details, such as the files a scan reads, the filters pushed to it, how a shuffle partitions the rows and which join Spark chose. With adaptive query execution on, the plan can still change while it runs.

## Sources

- [Jobs system table reference](https://docs.databricks.com/aws/en/admin/system-tables/jobs)
- [Query profile](https://docs.databricks.com/aws/en/sql/user/queries/query-profile)
- [Serverless compute limitations](https://docs.databricks.com/aws/en/compute/serverless/limitations)
- [pyspark.sql.DataFrame.explain](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html)
- [Performance Tuning](https://spark.apache.org/docs/latest/sql-performance-tuning.html)
- [Web UI](https://spark.apache.org/docs/latest/web-ui.html)