Skip to content
FR

Monitoring Jobs and Troubleshooting Spark Workloads

Step 23 of 30

Monitoring Jobs and Troubleshooting Spark Workloads

Introduction

This lesson covers the Troubleshooting, Monitoring, and Optimization section of the official exam outline (about 10% of the scored questions, shared with the next lesson on data layout). A pipeline that works on the day you build it will eventually be late, slow, or broken: a source sends a malformed file, a key becomes hot, a library is upgraded, a cluster runs out of memory. Data engineers are judged on how quickly they can find out that something is wrong, find why, and fix it without making things worse.

This lesson follows that sequence. First you learn to read Lakeflow Jobs run history, job statuses and the task graph to detect problems and to compare a run with its historical baseline. Then you learn to repair and rerun only what failed. Finally you open the Spark UI to recognize the classic performance problems (data skew, shuffle cost and disk spill), understand what adaptive query execution (AQE) does for you, and diagnose startup failures, library conflicts and out-of-memory errors.

Learning objectives

  • Use the run history of a job to identify failures, long runs and trends against a baseline.
  • Read the task graph to find the failed task and the upstream blockers of a skipped task.
  • Use repair run to rerun failed and dependent tasks without rerunning successful ones.
  • Read Spark UI stage and task metrics to recognize data skew, heavy shuffle and disk spill.
  • Explain what adaptive query execution does, including skew join handling.
  • Diagnose cluster startup failures, library conflicts and out-of-memory (OOM) errors and choose a fix.

1. Monitoring Lakeflow Jobs

Run history and job statuses

Every run of a job is recorded. The Runs tab of a job lists each run with its start time, duration, trigger type (scheduled, file arrival, manual, and so on) and status. The statuses you should recognize:

Status Meaning
Pending / Queued The run is waiting for compute or for a free concurrency slot
Running At least one task is executing
Succeeded All tasks succeeded (or were allowed to be skipped by design)
Failed A task failed and its retries are exhausted
Skipped A task did not run because a dependency condition was not met
Canceled / Timed out A person or the platform stopped the run; a timeout was reached
Upstream failed A task did not run because a task it depends on failed

A task can also show Excluded or Skipped when a run-if condition or an if/else condition task decided not to execute it. That is normal behavior, not an error, but it must be understood when reading a run.

The job page also offers a matrix view that shows many runs at once: each column is a run, each row a task. At a glance you see whether one task fails repeatedly (a code or data problem in that task) or all tasks failed in the same run (probably a platform, credential or permission problem).

The task graph and upstream blockers

A run opens as a graph of its tasks. Colors show each task's status. When a task is in "upstream failed" or "skipped", the graph helps you walk backwards to find the blocker: the first failed task with no failed ancestor. Fixing a skipped task is pointless; the root cause is the failed upstream task.

Concrete example. In a nightly job daily_sales, the tasks are ingest_orders -> clean_orders -> build_gold -> refresh_dashboard. The run shows ingest_orders green, clean_orders red, and the last two grey with "upstream failed". The engineer opens only clean_orders (the blocker), reads the error output, and ignores the grey tasks.

Historical baselines

A single run tells you if it failed. The history tells you if it is getting worse. Compare the duration and outcome of a run with the typical range from previous runs:

  • A job that normally takes 40 minutes and now takes 95 is a problem even though it succeeded. Likely causes: data volume growth, a skewed key, a changed query plan, a smaller cluster, or a slow upstream source.
  • A gradual increase over weeks suggests growth or missing maintenance (for example too many small files). A sudden jump on a given day suggests a change: new code, a new library version, a schema change.
  • A rising failure rate (for example three failures in the last ten runs) points to instability even if each failure was repaired.

You can also set a duration warning threshold and a timeout on tasks and jobs, so that an unusually long run raises a notification before it breaks a deadline, and a hung run is stopped instead of burning compute forever.

Notifications and system tables

Jobs can send notifications (email, Slack, Microsoft Teams, webhooks, PagerDuty) on start, success, failure, duration warning and streaming backlog events, at the job or task level. For monitoring across many jobs, Databricks publishes system tables in Unity Catalog (for example the Lakeflow tables for job runs and the billing usage table). With SQL you can build a dashboard of failure rates, average durations and cost per job. System tables are governed like other tables, and access has to be granted.

Retries

Each task can have retries (number of attempts and the minimum interval between them). Retries are useful for transient errors (a short network problem, a busy source) and wasteful for deterministic errors (a syntax error fails three times in a row). Set retries on tasks that call external systems; do not use retries to hide a bug.

2. Repairing and rerunning failed runs

When a run fails, you usually do not want to rerun everything. Repair run reruns only the failed tasks and the tasks that depend on them, keeping the results of tasks that succeeded. Repair keeps the same run (it adds a repair attempt to it) and you can change task parameters for the repair.

Action What it does Use when
Repair run Reruns failed and skipped/dependent tasks of that run; successful tasks are kept A task failed due to a transient or fixed cause; earlier tasks were fine and expensive
Run now (new run) Starts a new full run of the job You want a complete rerun, or a different parameter set
Run now with different parameters Same as above with overrides Backfills or reprocessing a specific date

Repair is only safe if the tasks are idempotent: running a task again must produce the same result and not duplicate data. This is why earlier lessons insisted on MERGE, COPY INTO and checkpoints instead of blind appends.

Concrete example. The build_gold task of a 3-hour job fails at 02:10 because a service principal's permission on a schema was revoked by mistake. The ingestion and cleaning took 2 hours 40 minutes. After the permission is restored, the engineer clicks Repair run, which runs only build_gold and refresh_dashboard, finishing in 20 minutes instead of repeating 3 hours of work.

3. Reading the Spark UI

For a task in a notebook or a job, the Spark UI shows how Spark executed the work. You do not need to know every tab. For the exam, remember the hierarchy:

  • A job (Spark) is triggered by an action such as a write.
  • A job is split into stages at shuffle boundaries.
  • A stage is a set of parallel tasks, one per partition.

Open the slowest stage and look at the task metrics summary: minimum, 25th percentile, median, 75th percentile and maximum for duration, input size, shuffle read size and spill. The shape of these numbers tells the story.

Data skew

Data skew means that the data is unevenly distributed across partitions. After a shuffle (a join, groupBy or window), all rows with the same key go to the same task. If one key (for example customer_id = NULL, country = 'US', or one giant account) holds a large fraction of the rows, one task receives far more data than the others.

Symptoms in the task summary:

  • The maximum task duration is many times the median (for example median 8 seconds, max 14 minutes).
  • The maximum shuffle read size is far larger than the median.
  • In the event timeline, most tasks finish early and a few tasks run alone for a long time while the rest of the cluster is idle.
  • The stage seems "stuck at 199 of 200 tasks".

Fixes, from least to most effort:

  1. Filter or fix the hot value if it is a data-quality problem (for example NULL keys or a default value like -1 that should not be joined).
  2. Let AQE handle it (see below). AQE is enabled by default in current runtimes.
  3. Broadcast the small side of a join if it is small enough, which avoids shuffling the large side at all.
  4. Salt the key: add a random suffix to spread the hot key across tasks, then aggregate again. This is a last resort because it complicates the code.

Shuffle

A shuffle moves data across the network between stages. It is expensive because it writes intermediate data to disk, sends it across executors, and sorts or hashes it. Signs of a heavy shuffle: large Shuffle Write and Shuffle Read values, long stage durations with a lot of network and disk activity.

Reduce shuffle by filtering and selecting columns before joins and aggregations, by broadcasting small tables, and by not repeating wide operations (groupBy, join, distinct, orderBy) unnecessarily. The number of shuffle partitions is controlled by spark.sql.shuffle.partitions (200 by default), which you met in the tuning lesson, and AQE can coalesce small partitions automatically.

Disk spill

Spill happens when a task does not have enough execution memory to hold its data (for a sort, aggregation, or join), and Spark writes the excess to disk. The Spark UI reports Spill (Memory) and Spill (Disk) per task. Spill is slower than memory, though it avoids a crash. Causes and fixes:

Cause Fix
Too few, too large shuffle partitions Increase the number of shuffle partitions, or let AQE manage it
Data skew: one task gets too much data Treat the skew first
Nodes with too little memory per core Use a memory-optimized instance type
Reading too many columns/rows Filter and project early

If spill appears in all tasks, partitions are too big or memory is too small. If it appears in only a few tasks, suspect skew.

Reading the three symptoms together

Observation Probable cause
Max duration >> median, a few huge tasks Data skew
Very large shuffle read/write Wide transformation on lots of data; a join that could be broadcast
Spill (disk) in many tasks Partitions too big for memory
Thousands of tiny tasks of a few milliseconds Too many small partitions or too many small files

4. Adaptive query execution (AQE)

Adaptive query execution re-optimizes a query while it runs, using real statistics collected at the end of each shuffle stage. It is enabled by default in current Databricks runtimes and serverless compute. It does three main things:

  1. Coalesces shuffle partitions. If after a shuffle the partitions turn out small, AQE merges them into fewer, larger ones. You no longer need to hand-tune spark.sql.shuffle.partitions for every query.
  2. Switches join strategy. If, at runtime, one side of a sort-merge join turns out small enough, AQE converts it to a broadcast join.
  3. Handles skew joins. AQE detects skewed partitions in a sort-merge join (a partition much larger than the median and larger than a size threshold) and splits them into smaller sub-partitions, so that several tasks share the hot key's work. In the Spark UI the join node shows that skew handling was applied.

AQE does not fix everything: it cannot help with skew in an aggregation that has no splittable work, with a key that is so hot that even a split part is huge, or with a bad query that reads too much data. It works on plans with shuffles, and some situations (for example streaming queries) do not use it in the same way.

Concrete example. A join between orders (2 billion rows) and customers (80 million rows) on customer_id slows down after a marketing campaign loads 300 million orders for a single guest-checkout customer_id = 0. The Spark UI shows one task reading 40 GB of shuffle data while the median is 300 MB. With AQE on, the skewed partition is split and the stage finishes in minutes. If the engineer sees AQE disabled in a configuration (for example spark.sql.adaptive.enabled = false set in a cluster setting years ago), the first fix is to remove that override.

5. Cluster, library and memory problems

Cluster startup failures

When compute does not start, the cluster event log or the job run page shows a termination reason. Common causes:

  • Cloud capacity or quota: the cloud account has no more vCPU quota or the instance type is unavailable in the zone. Fix: request quota, choose another instance type or zone, or use serverless compute.
  • Permissions and networking: a missing role, a blocked subnet, an unavailable security group or a missing network route to the control plane.
  • Invalid configuration: an unsupported runtime version, an invalid init script path, an invalid Spark configuration.
  • Init script failure: an init script returns a non-zero exit code, so the cluster fails to start. Fix the script, which is shown in the init script logs.

Pools and serverless compute reduce startup time and make capacity problems less likely.

Library conflicts

A library conflict occurs when two libraries need incompatible versions of the same dependency, or when a library you install overrides a package that the runtime already provides. Typical symptoms: ImportError, AttributeError: module has no attribute, NoSuchMethodError for Java/Scala libraries, or a job that worked before a library upgrade and fails after it.

Fixes:

  1. Pin versions in requirements.txt or in the bundle/job library definition so that the environment is reproducible.
  2. Prefer job-scoped or notebook-scoped libraries over cluster-wide installs to limit blast radius.
  3. Use the Databricks Runtime release notes to know which versions are already installed, and avoid reinstalling them with different versions.
  4. Reproduce the problem in a clean, separate environment, then change one library at a time.

Out-of-memory errors

OOM errors exist in two places:

Where Typical message Common causes Fixes
Driver java.lang.OutOfMemoryError on the driver, "Driver is up but is not responsive", collect() of a large DataFrame collect(), toPandas(), or display() on a huge dataset; large broadcast; too many small files making a large file listing Avoid collecting large data to the driver; aggregate or limit first; use a larger driver node
Executor ExecutorLostFailure, "Container killed by YARN/OOM", task failures with spill and then loss Skewed partitions, very large partitions, memory-heavy UDFs, too many cores per executor for the memory Reduce partition size (more partitions), fix skew, use memory-optimized nodes, avoid pandas UDFs on huge groups

Increasing spark.executor.memory or spark.driver.memory is sometimes the right answer, but on Databricks the memory is tied to the node type, so the usual fix is a larger or memory-optimized node, or removing the cause (collecting too much, skew).

Worked example: diagnosing a slow nightly job

Scenario. The job daily_customer_360 normally runs in 35 minutes. For three days it has taken more than 2 hours but still succeeds. The run history shows a duration jump on Tuesday. The team notes that on Monday a new marketing source was added that creates many records without a customer id (stored as NULL). The Spark UI for the slowest stage shows: median task duration 6 seconds, 75th percentile 9 seconds, maximum 1 hour 40 minutes; shuffle read max 52 GB versus median 400 MB; spill (disk) only in the largest task.

Step 1: Compare with the baseline. A sudden jump on a specific day points to a change in the data or the code, not slow growth.

Step 2: Locate the slow stage. Open the Spark UI for the longest task of the run and sort stages by duration.

Step 3: Read the task metrics. A maximum far above the 75th percentile, a huge shuffle read in one task, and spill only in that task all point to data skew, not to a generally undersized cluster (which would show spill in many tasks).

Step 4: Identify the key. Group the join key and count rows: customer_id IS NULL holds 380 million rows. NULL keys never match in an inner join, yet they are still sent to one partition by the shuffle.

Step 5: Fix. Filter the NULL keys out of the join input (or route them to a separate quarantine table) before the join. Check that AQE skew handling is enabled for any remaining skew.

Step 6: Re-measure. Run the job again and compare duration and max/median task times with the baseline.

Why the alternatives are worse. Doubling the cluster size would cost more and not help, because a single task still processes the hot partition. Increasing spark.sql.shuffle.partitions does not split a single key. Adding retries would just repeat a slow run.

Common pitfalls and exam tips

  • Find the first failed task. "Upstream failed" and "skipped" tasks are consequences, not causes.
  • Repair run reruns only failed and dependent tasks and requires idempotent tasks; a new run repeats everything.
  • A job can succeed and still be a problem. Compare duration to the historical baseline.
  • Skew = one or a few very slow tasks; spill in all tasks = undersized partitions or memory.
  • Do not "fix" skew by adding nodes. Fix the key distribution, broadcast, or rely on AQE.
  • AQE is on by default and handles partition coalescing, join strategy switching and skew join splitting.
  • Library conflicts: pin versions and scope libraries to the job.
  • Driver OOM usually comes from collect(), toPandas() or huge broadcasts; executor OOM from huge or skewed partitions.
  • Retries are for transient errors, not for bugs.

Exam tip: The question often gives you a few numbers from the Spark UI. If maximum is far above the median for duration or shuffle read, answer skew. If spill is everywhere, answer partition size or memory. If the symptom is "run takes longer than usual but succeeds", answer comparison with historical runs.

Mini case study

Situation. A weekly job inventory_snapshot failed at 05:00. The run page shows five tasks: extract_erp (green), load_stock (red, error ModuleNotFoundError: No module named 'pyarrow.compute'), build_aggregates (grey, upstream failed), quality_checks (grey) and publish (grey). The job had been green for months. Yesterday, a colleague added pyarrow==3.0.0 to the cluster's libraries for another notebook. The load_stock task takes 2 hours; extract_erp takes 90 minutes.

Task. Identify the cause, fix the issue and recover with minimum rework.

Solution.

  1. The blocker is load_stock; the grey tasks are consequences.
  2. The error and the recent change point to a library conflict: an old pinned pyarrow overrides the newer version provided by the runtime.
  3. Remove the pin from the shared cluster, and define libraries at the job level with explicit, compatible versions.
  4. Use Repair run so extract_erp (90 minutes) is not repeated: only load_stock and the dependent tasks run.
  5. Add a notification on failure and a task timeout, and keep the dependency list under version control in the bundle.

Key terms

  • Run history: the list of previous runs of a job with status, duration and trigger.
  • Task graph: the visual dependency graph of a run.
  • Upstream blocker: the first failed task that causes dependent tasks to be skipped.
  • Repair run: rerun of failed and dependent tasks of the same run.
  • Baseline: the typical duration and outcome from past runs.
  • Spark UI: web interface with jobs, stages, tasks and SQL plans.
  • Data skew: uneven distribution of data across partitions.
  • Shuffle: redistribution of data across the network between stages.
  • Spill: writing intermediate data to disk when memory is insufficient.
  • Adaptive query execution (AQE): runtime re-optimization (partition coalescing, join switching, skew join handling).
  • OOM: out-of-memory error on the driver or an executor.

Summary checklist

  • I can read a run's task graph and find the upstream blocker.
  • I can compare a run with its historical baseline and spot a trend.
  • I know when to use repair run and when to start a new run.
  • I can recognize skew, heavy shuffle and spill in Spark UI metrics.
  • I can describe what AQE does and that it is enabled by default.
  • I can diagnose startup failures, library conflicts and OOM errors, and pick a fix.

Official sources