Monitoring, Troubleshooting and Optimizing Pipelines: Lakeflow Jobs Run History, Spark UI, Liquid Clustering and Predictive Optimization (Databricks Data Engineer Associate)

Pipeline health in Lakeflow Jobs is read from run and task statuses, and the fastest workspace-wide view of them is the Runs tab under Jobs & Pipelines. It lists the job and pipeline runs of every job in the workspace for the last 60 days, so an on-call engineer can triage many failures without opening jobs one by one or writing SQL.

Statuses you need to read correctly

The whole run's status follows its leaf tasks (Run if rules and repair are covered in the Lakeflow Jobs control-flow lesson). For troubleshooting, the key habit is that Upstream failed tasks hold no error of their own; their output will not explain anything.

What the Runs tab gives you

• A finished-runs graph of successes and failures, covering the last 48 hours by default; the time filter matches completed runs on their end time. Non-admins choose Run as: me to see it.
• A Top 5 error types table for the range, which tells you whether a wave of failures shares one cause.
• A runs list with filters: status (for example Failed), time range, error code, and the run-as identity (a user or a service principal).

Pick the filter that matches the question. To isolate failures caused by one identity losing access, filter by run-as: the job owner is a different property, and an error-code filter both pulls in unrelated jobs and misses that identity's runs that failed with a different code. To review every failure of the week, filter by Failed status and widen the time range to the week.

To compare current execution times against historical baselines, open the job's Runs tab and use the matrix view. It lines up recent runs as columns: the Run total duration row shows each run's duration and state, and every task has its own row, so a task whose duration has been climbing stands out immediately.

The runs list confirms that a job slowed down; it does not split the growth by task. Sorting it by duration and opening the slowest run still leaves you looking at one run.

Warning versus Timeout

A job's Run duration metric threshold has two independent values. The Warning is an expected completion time: crossing it raises a duration-warning event that can trigger a notification, while the run keeps going. The Timeout stops the run when it is reached and marks it Timed out. Tasks can have their own timeouts too, which stop the task and leave dependants Upstream failed.

So a run with a 30-minute Warning and a 60-minute Timeout that needs 75 minutes raises a warning at 30 minutes, keeps running, and ends Timed out at 60 minutes; the earlier warning is not withdrawn. When a slow run must be allowed to finish, use the Warning alone.

Beyond the UI: the jobs system tables

The Jobs UI keeps run history for 60 days. For older baselines, account-wide dashboards or failure-rate percentiles, query the system.lakeflow schema, which keeps up to 365 days once enabled:

• job_run_timeline and job_task_run_timeline record each job run and task run, with start and end times and the result state. These are the tables for durations and failure rates.
• jobs and job_tasks record job and task definitions as slowly changing (SCD2) rows. They show how a job was configured over time, not how long it ran or whether it failed.

A system-table query refreshes automatically when scheduled behind a dashboard, which a manual export of the runs list never will.

Inside a single job run, the Graph view is the tool for spotting upstream blockers: it draws the run's task DAG and colours each task by status, so you follow the edges back from a waiting or Upstream failed task to the task that is still running or marked Failed.

• Graph view: dependencies and per-task status. Use it to answer "what is this task waiting on?" and "where did the failure start?"
• Timeline view: every task on a time axis. Use it to see which task stretched the end-to-end time and whether tasks meant to run in parallel actually overlapped.
• List view: tasks with status and duration in a table. Sorting by duration finds the slowest task, but it cannot show when tasks ran relative to each other.

Keep the levels apart. Lakeflow job tasks are not Spark tasks: the Spark UI's Jobs and Stages tabs describe the Spark work of one execution and say nothing about job task dependencies. The job's Tasks tab edits the definition, and the schedule and trigger settings only decide when runs start.

The Spark UI exposes the three classic bottlenecks named in the exam guide through stage-level metrics: shuffle appears as Shuffle Write and Shuffle Read between stages, skew as one task far slower and larger than its peers in the stage's Summary Metrics, and spill as spill statistics on the stage page.

Shuffle: where stages begin and end

Wide transformations (joins, groupBy, distinct, window partitions, repartition) need all rows with the same key together, so Spark writes Shuffle Write data at the end of one stage and reads it as Shuffle Read at the start of the next. Narrow transformations (filter, select, withColumn) work inside each partition and stay in one stage. Reading source files shows as Input; writing a table shows as Output; neither is a shuffle.

In the SQL / DataFrame tab's query plan, the shuffle is an Exchange node, for example Exchange hashpartitioning(order_id#7, 200): rows are repartitioned by key across executors and a new stage begins. FileScan, Filter and Project nodes move no data between executors. (Photon plans use Photon-named operators, but the idea is the same.)

Skew: compare Max with the 75th percentile

Each stage page has a Summary Metrics table with Min, 25th percentile, Median, 75th percentile and Max for task duration, shuffle read size/records, spill and more. Its Peak Execution Memory row shows the most execution memory each task used; read it next to the spill rows, since tasks that spill are the ones whose peak hit the memory available to them. The quick skew test from Databricks' Spark UI guide: if the Max task duration is far above the 75th percentile (around 50% higher or more), suspect skew; in a healthy stage the two are close. A few unusually fast tasks (a small Min) do not matter, because the slowest task decides when the stage ends.

Confirm the cause with the data columns. True key skew means the slow task also read far more data: its Shuffle Read Size / Records at Max is far above the 75th percentile. If the slowest task read a normal amount of data, it is not key skew, and salting will not help; look at that task's executor instead (garbage collection, a slow or failing node).

Patterns that look similar but are not skew: too many shuffle partitions produces lots of tiny tasks with scheduling overhead; a retried task after a lost executor reprocesses a normal-sized partition; a large scheduler delay points at waiting for resources.

Spill means a task's data did not fit in execution memory and was written to disk, which slows the stage. Spill statistics appear in the stage details at the top of the stage page only when the stage spilled; if they are absent, the stage did not spill. Other tabs are not spill views: the Storage tab shows cached data, the Executors tab's storage memory is for cached blocks, and the Environment tab lists configuration.

Spill (Memory) versus Spill (Disk)

The two figures describe the same spilled data measured twice. Spill (Memory) is its deserialized size while it was in memory; Spill (Disk) is its serialized, usually compressed, size once written to disk. That is why the memory figure is normally the larger one. Neither figure is data "kept safely in memory", and they are not a split between shuffle reads and writes.

Where the spill sits decides the fix

For even spill, adding more workers of the same type does not help: each task still gets the same execution memory for a partition of the same size. A worker type with more memory per core does, and needs no code or Spark configuration change. For skew-driven spill, more memory leaves the one huge task dominating the stage, and fewer shuffle partitions makes every partition bigger without separating the hot key.

Skew is fixed by spreading the oversized partition's rows over more tasks; which tool fits depends on the operation and on what you are allowed to change.

Adaptive query execution (AQE) skew join

With AQE on, Spark splits skewed partitions in sort-merge and shuffle-hash joins at runtime. A partition counts as skewed only when it exceeds both spark.sql.adaptive.skewJoin.skewedPartitionFactor times the median partition size (default 5) and spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256 MB). A 200 MB partition that is 30 times the median therefore is not split until the byte threshold is lowered below its size; raising the factor makes detection stricter, not looser. AQE skew handling does not apply to aggregations (groupBy) or broadcast joins.

Salting and two-stage aggregation

When AQE cannot help, for example a groupBy on a hot key, add a random salt to the key, aggregate by key plus salt, then aggregate again by the key alone. The hot key's rows are spread over many tasks in the first pass.

NULL join keys

All NULL keys hash to the same partition, so a column full of NULLs creates a hot partition that more shuffle partitions cannot break up. NULLs never match in an equi-join, so in a left join you can join only the rows with a key and union the NULL-key rows back with null columns. This keeps every row and removes the hot partition; simply filtering NULLs out would drop rows the output needs.

The REBALANCE hint for skewed writes

When the stage that writes a table has a few huge tasks and many tiny output files, add a REBALANCE hint to the query that produces the rows. With AQE on, it evens out output partition sizes, splitting large ones and combining small ones. COALESCE only merges partitions, BROADCAST changes a join, and turning AQE off removes the mechanism.

Shuffling less data

The biggest shuffle win is usually in the code: aggregate or filter large inputs before the join, so far fewer rows cross the Exchange. Repartitioning by the join key is itself a full shuffle, and caching avoids rereading files but still shuffles every row. Shuffle partition counts and broadcast thresholds are covered in the Spark tuning lesson of the transformation domain.

Liquid Clustering is the Delta Lake data layout that co-locates rows by up to four clustering keys for data skipping, and lets you change those keys later without rewriting existing data. Databricks recommends it for new tables instead of partitioning and ZORDER.

Turning it on and changing keys

• CREATE TABLE ... CLUSTER BY (col1, col2) sets keys at creation (also available in CTAS and the DataFrame writer's clusterBy).
• ALTER TABLE t CLUSTER BY (new_col) enables clustering on an existing unpartitioned Delta table or changes its keys. CLUSTER BY NONE turns clustering off.
• Key order does not matter: clustering works on the set of keys.

OPTIMIZE versus OPTIMIZE FULL

Clustering is applied by OPTIMIZE, which is incremental: it clusters data that has not yet been clustered. Changing keys affects new clustering work only. To recluster all existing records by the new keys in one operation, run OPTIMIZE t FULL. ANALYZE TABLE only collects statistics, and REORG TABLE ... APPLY (PURGE) rewrites files to purge soft-deleted data.

What it does not combine with

A table cannot be both partitioned and liquid clustered, and Z-ordering cannot be used on a clustered table. Plain ALTER TABLE ... CLUSTER BY is for unpartitioned tables and fails on a partitioned one. A partitioned table is converted either in place with ALTER TABLE t REPLACE PARTITIONED BY WITH CLUSTER BY (cols) (or AUTO), available on Databricks Runtime 18.1 and above for managed and external tables, or by rebuilding it without partitions (for example with CTAS). Moving off partitions is the right call when a partition column is high-cardinality or uneven, for example a handful of very large regions next to hundreds of near-empty ones, which creates both oversized and small-file partitions that compaction or Z-ordering inside partitions cannot fix.

Automatic liquid clustering

CLUSTER BY AUTO (at create time or through ALTER TABLE) lets Databricks choose the clustering keys from the table's query history and change them when the expected data-skipping gains outweigh the cost of reclustering. It requires Unity Catalog managed tables with predictive optimization enabled. An external table can still be liquid clustered, but only with keys you set yourself. Predictive optimization never picks keys for a table without CLUSTER BY AUTO.

Predictive optimization is a Databricks feature that decides which Unity Catalog managed tables benefit from maintenance and runs OPTIMIZE, VACUUM and ANALYZE on them automatically, on serverless compute, so you no longer schedule a maintenance job that misses new tables and wastes work on tables that do not need it.

Scope and inheritance

• It runs only on Unity Catalog managed tables. External tables are never maintained by it, whatever the setting, so they still need a scheduled job that runs OPTIMIZE and VACUUM. Shortening a retention property does not run VACUUM for you.
• It can be enabled for an account, a catalog, a schema or a single table. An account admin sets the account-level default; catalogs, schemas and tables then inherit from their parent unless overridden: ALTER CATALOG c ENABLE | DISABLE | INHERIT PREDICTIVE OPTIMIZATION and ALTER SCHEMA c.s ENABLE | DISABLE | INHERIT PREDICTIVE OPTIMIZATION.
• Set it at the lowest level you mean. Enabling one schema overrides the inherited setting for that schema's managed tables only; enabling the catalog would switch on every schema that inherits from it; INHERIT just takes the parent's value.

How it relates to nearby features

Checking what it did

The system table system.storage.predictive_optimization_operations_history records every operation it ran: table, operation type (compaction, vacuum, clustering, statistics), time and metrics, so one query covers a whole catalog. DESCRIBE HISTORY works one table at a time, and system.billing.usage shows what the feature consumed, not what it did to each table. Predictive optimization does not run as Lakeflow jobs, so it never appears in job run tables.

When classic compute fails to start, the first place to look is the compute event log, which records the failed start and its reason; the Spark UI cannot help because Spark never started, and job run history only tells you that the run failed.

Common startup causes

The event log records which init scripts started and finished, not their output. To see why a script failed, configure cluster log delivery: init script output is then written under <log-path>/<cluster-id>/init_scripts and survives termination. Removing the script and starting the cluster produces no log of the failing command. A runtime version change, a longer job timeout or an init script that "retries" cannot create cloud capacity that does not exist.

Library conflicts

Python libraries come from three layers, and the higher one wins for a notebook: notebook-scoped (%pip install) over compute-scoped (installed on the cluster's Libraries tab) over the libraries bundled in the Databricks Runtime.

• Need a different version in one notebook on a shared cluster? %pip install pkg==x.y.z in that notebook, then dbutils.library.restartPython(), because %pip does not restart the Python process and an already-imported version stays loaded. Other notebooks keep the compute-scoped version.
• Import fails after adding a compute-scoped library? Check the compute's Libraries tab, which shows each library's version, source and install status, including failed installs.
• Swapping a compute-scoped version? An uninstalled library shows Uninstall pending restart and stays loaded until the cluster restarts; detaching and reattaching notebooks is not enough.
• Job broke with no code change? A PyPI library listed by name only installs the latest release on every new job cluster. Pin versions for repeatable jobs.
• Databricks does not recommend installing libraries through init scripts, and a second compute-scoped version of the same package conflicts for every notebook on the cluster.

Out-of-memory failures come from either the driver, which coordinates the job and receives any data you bring back to it, or the executors, which process partitions; the symptoms differ and so do the fixes.

Driver OOM: stop pulling data to one machine

collect() and toPandas() gather every row into the driver. Repartitioning first changes nothing, because all partitions still end up in one place, and raising spark.driver.maxResultSize only lifts a cap; the driver still has to hold everything. If a bigger driver is not an option, rewrite the work with DataFrame functions and write results with df.write, so executors do the work and the driver never holds the data.

Executor OOM: rule out lost nodes first

An executor can also disappear because autoscaling removed it or a spot instance was reclaimed. Check the compute event log for resize and instance-loss events during the stage before concluding that memory ran out.

A classic real memory cause is a stale /*+ BROADCAST(t) */ hint: a broadcast copies the whole table to every executor, so when the table grows to gigabytes the join stage loses executors. An explicit hint is honoured regardless of spark.sql.autoBroadcastJoinThreshold, more workers only add more copies, and more shuffle partitions do not split a broadcast table. Remove the hint so the planner can choose a shuffle join.