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

Part 1 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 for the Databricks queries, and module 3's bigger shop 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:

-- 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, 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:

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:

== 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 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:

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 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 turns the time a run takes into what it costs, from the shop job's own runs.