Spark Bottlenecks, Table Layout and Compute Failures

What it is

When a job is slow or fails, the cause is usually one of a few patterns: uneven data across tasks, too much data moved between nodes, not enough memory per task, a poor table layout, or compute that cannot start or run the code.

The Spark UI shows stage and task metrics for slow queries. Liquid Clustering and predictive optimization keep Delta tables laid out well. The compute event log, library status and driver logs explain startup, library and memory failures.

How it works

Reading stage metrics in the Spark UI

Open the Spark UI from the cluster or the query, find the slowest stage, and open its detail page. The Summary Metrics table shows task duration, input, shuffle read and write and spill at Min, 25th percentile, Median, 75th percentile and Max.

Adaptive Query Execution splits skewed partitions in joins, not in groupBy aggregations. For a skewed aggregation, add a random salt to the key, aggregate, then aggregate again without it.

from pyspark.sql import functions as F
salted = df.withColumn("salt", (F.rand() * 16).cast("int"))
partial = salted.groupBy("customer_id", "salt").agg(F.sum("amount").alias("s"))
result = partial.groupBy("customer_id").agg(F.sum("s").alias("amount"))

Liquid Clustering

Liquid Clustering organizes a Delta table's data by up to four clustering keys, chosen from columns used in filters. Databricks recommends it for all new tables, including streaming tables and materialized views.

CREATE TABLE sales.orders (order_id BIGINT, customer_id BIGINT, order_date DATE)
CLUSTER BY (customer_id, order_date);
ALTER TABLE sales.orders CLUSTER BY (order_date);
OPTIMIZE sales.orders;

Clustering is incremental: OPTIMIZE rewrites only data that is not yet clustered. Changing keys with ALTER TABLE ... CLUSTER BY does not rewrite existing data; OPTIMIZE ... FULL reclusters it.

CLUSTER BY AUTO lets Databricks choose the keys from the query workload; it needs predictive optimization and a Unity Catalog managed table. Liquid Clustering cannot be combined with PARTITIONED BY or ZORDER.

Predictive optimization

Predictive optimization runs OPTIMIZE, VACUUM and ANALYZE automatically on serverless compute, when Databricks judges they are worth the cost. It covers Unity Catalog managed tables only, never external tables.

ALTER CATALOG sales ENABLE PREDICTIVE OPTIMIZATION;
ALTER SCHEMA sales.staging INHERIT PREDICTIVE OPTIMIZATION;

It is set at account, catalog, schema or table level, and lower levels inherit by default. Its operations are recorded in system.storage.predictive_optimization_operations_history.

Cluster startup failures

If a cluster never reaches running, the Spark UI has nothing to show. Open the compute's event log, which lists each step and the error that stopped it.

Fixes include another instance type or zone, a quota increase, correcting the init script, or serverless compute where the workload allows it.

Library conflicts

A library conflict happens when an installed package needs a different version of a package that the Databricks Runtime already includes, or when two libraries need incompatible versions. The compute's Libraries tab shows a failed install; imports fail or behave oddly.

Notebook-scoped libraries installed with %pip install affect only that notebook session. Pin versions that match the runtime, then restart Python with dbutils.library.restartPython().

Out of memory errors

A driver out of memory error typically shows as the Spark driver stopping unexpectedly and restarting. Executor problems show as failed tasks and lost executors in the Spark UI.

How the exam asks about this

• "In one stage, the max task duration is 40 times the median." Data skew; for a join rely on AQE skew handling, for an aggregation salt the key.
• "Queries filter on customer_id and the table is not partitioned; reduce files scanned with low upkeep." Liquid Clustering with CLUSTER BY (customer_id).
• "The cluster stays in pending and then terminates." Read the compute event log for the cloud or init script error.