Optimizing Databricks Compute Costs and Spark Workloads

Databricks spend rarely comes from one bad decision. It accumulates from defaults nobody revisited: scheduled jobs on all-purpose clusters, a SQL warehouse sized for a Monday-morning peak, tables made of small files, a streaming query polling an empty source around the clock. Each is cheap to fix once you can see it.

This article is a working method for engineers who own a Databricks bill: build a cost baseline from system tables, choose compute per workload, tune the Spark workloads that dominate spend, keep Delta table layout healthy, control streaming cost, and govern it so the savings last. Examples assume Databricks on AWS and use generic names such as main.sales.orders.

Establish a cost baseline before tuning anything

Tuning without attribution optimizes whatever is in front of you rather than whatever costs the most. Start with a dataset that answers, for any day, which team, pipeline and product consumed what, at list price, on which kind of compute.

What the billing system tables give you

system.billing.usage has one row per usage record: quantity and unit (DBUs for compute), sku_name, billing_origin_product, a usage_metadata struct with attribution ids (job_id, job_run_id, warehouse_id, cluster_id), identity_metadata.run_as, custom_tags, and product_features flags such as is_serverless. system.billing.list_prices holds each SKU’s list price over time, so cost is a join on the SKU and the price window that contains the usage.

Two details decide whether the numbers are right. Corrections arrive as extra rows, a RETRACTION with a negative quantity followed by a RESTATEMENT, so sum every record type rather than filtering to ORIGINAL. And this is list price for the Databricks side only. It ignores negotiated discounts, and classic compute also incurs cloud provider charges for instances and storage, which you attribute from AWS cost data using the same tags.

-- List-price cost per team and product over the last 30 days.
-- RETRACTION rows carry negative quantities, so summing all record types
-- yields corrected usage.
SELECT
  COALESCE(u.custom_tags['team'], 'untagged')                      AS team,
  u.billing_origin_product                                         AS product,
  COALESCE(u.product_features.is_serverless, false)                AS serverless,
  ROUND(SUM(u.usage_quantity), 1)                                  AS dbus,
  ROUND(SUM(u.usage_quantity * p.pricing.effective_list.default), 2) AS list_cost
FROM system.billing.usage AS u
JOIN system.billing.list_prices AS p
  ON  p.sku_name = u.sku_name
  AND u.usage_end_time >= p.price_start_time
  AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
WHERE u.usage_date >= date_sub(current_date(), 30)
GROUP BY ALL
ORDER BY list_cost DESC;

Attribution with tags, usage policies and job identity

Classic compute carries default tags (ClusterId, ClusterName, Creator, plus JobId and RunName on job compute) and any custom tags set on the cluster, pool or job. Custom tags reach both custom_tags in the usage table and the EC2 and EBS resources. Serverless has no cluster to tag, so attribution comes from serverless usage policies (introduced as serverless budget policies): named sets of tags applied to the serverless usage of the users, jobs, pipelines, notebooks and endpoints assigned to them. Policy tag changes apply only to usage that starts afterwards, so set policies up before moving workloads.

Keep the taxonomy small and enforced: three or four keys such as team, cost_center and env, applied by policy. Untagged spend needs an owner.

Shared SQL warehouses need an allocation model. system.query.history records executed_by, query_source and total_task_duration_ms, so split each warehouse’s cost by each consumer’s share of task time. Agree the model with finance; it is an approximation, not metered truth.

For jobs, join usage to the latest row per job in system.lakeflow.jobs, which keeps a row for every change:

-- Cost per job and per run over the last 30 days, at list price.
WITH job_names AS (
  SELECT workspace_id, job_id, name
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
job_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.job_id                                  AS job_id,
    COUNT(DISTINCT u.usage_metadata.job_run_id)              AS runs,
    SUM(u.usage_quantity * p.pricing.effective_list.default) AS list_cost
  FROM system.billing.usage AS u
  JOIN system.billing.list_prices AS p
    ON  p.sku_name = u.sku_name
    AND u.usage_end_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= date_sub(current_date(), 30)
  GROUP BY ALL
)
SELECT
  c.workspace_id,
  c.job_id,
  n.name                                       AS job_name,
  c.runs,
  ROUND(c.list_cost, 2)                        AS list_cost,
  ROUND(c.list_cost / NULLIF(c.runs, 0), 2)    AS list_cost_per_run
FROM job_cost AS c
LEFT JOIN job_names AS n
  ON n.workspace_id = c.workspace_id AND n.job_id = c.job_id
ORDER BY list_cost DESC
LIMIT 25;

job_id is not populated for jobs that run on all-purpose clusters. That gap is itself a finding: scheduled work on all-purpose compute appears under ALL_PURPOSE with no job identity, usually run as a service principal. Find it and move it.

Choose compute deliberately

Compute choice sets the unit price of everything that runs on it. Decide per workload, write the decision down, and revisit it when the workload changes.

Job compute for scheduled work

All-purpose clusters are built for interactive work and accumulate idle time between commands. Databricks’ cost guidance states that non-interactive workloads cost significantly less on job compute, which exists only for the duration of a run. Interactive clusters still need a short auto-termination setting, enforced by policy.

Serverless or classic

Serverless compute removes cluster sizing, start-up waits and idle capacity, and with them some levers: instance choice, spot, pools, init scripts, most Spark configurations, and the Spark UI (the query profile replaces it). Serverless jobs and pipelines run in performance optimized mode or in standard mode, which uses fewer DBUs in exchange for a start-up delay of several minutes and is unavailable for some task types, such as continuous pipelines. product_features.performance_target records the mode, so compare cost per run before and after switching.

SQL warehouses: size for the query, scale for concurrency

Size sets the resources for a single query; the cluster range sets how much concurrency the warehouse absorbs. Follow the documented rule: if queries spill to disk, increase the size; if they queue, raise the maximum number of clusters. Serverless warehouses start in seconds, so a short auto-stop costs little; pro and classic warehouses take minutes, so a very short auto-stop trades idle cost for cold starts. Find the statements worth fixing before resizing:

-- Warehouse statements that spilled to disk in the last 7 days, worst first.
SELECT
  statement_id,
  executed_by,
  compute.warehouse_id                           AS warehouse_id,
  total_duration_ms,
  total_task_duration_ms,
  ROUND(read_bytes / POWER(1024, 3), 1)          AS read_gib,
  ROUND(spilled_local_bytes / POWER(1024, 3), 1) AS spilled_gib,
  read_io_cache_percent
FROM system.query.history
WHERE start_time >= current_timestamp() - INTERVAL 7 DAYS
  AND compute.type = 'WAREHOUSE'
  AND spilled_local_bytes > 0
ORDER BY spilled_local_bytes DESC
LIMIT 20;

Pools, spot and fleet instances on AWS

Databricks charges no DBUs for idle pool instances, but AWS bills for them, so a pool’s idle minimum is a standing cost. For spot capacity, set aws_attributes.availability to SPOT_WITH_FALLBACK and aws_attributes.first_on_demand to 1 so the driver stays on demand. Work on a reclaimed worker can be lost and recomputed, so spot suits idempotent batch with short stages, not long stages or streaming. Fleet instance types and flexible node types reduce launch failures by falling back across instance types when capacity is short.

Photon: when it pays for itself

Photon is the default on SQL warehouses and serverless, and on by default for classic compute, where it is a real cost decision because Photon instance types consume DBUs at a different rate. It pays for itself when the runtime reduction outweighs the rate difference, most likely for long-running scans, joins and aggregations over large data. Queries that finish in a couple of seconds gain little, and Photon does not accelerate Python UDFs, RDD or Dataset APIs, or stateful streaming; those parts fall back to the standard engine.

Autoscaling behavior

Optimized autoscaling (Premium plan and above) can scale down a busy cluster by looking at shuffle file state, and job compute scales down sooner than all-purpose compute. The traps are a high minimum, paid for with or without work, and a generous maximum that lets an inefficient query buy its way out of a problem. Check system.compute.node_timeline first: workers at low CPU for weeks mean oversized compute. For streaming, autoscaling on classic compute has scale-down limitations; Databricks recommends Lakeflow pipelines (Spark Declarative Pipelines, formerly Delta Live Tables) with enhanced autoscaling.

Tune the workloads that dominate the bill

Work down the cost-per-run list, reading the query profile (serverless and SQL warehouses) or the Spark UI (classic) before changing anything: which operators take the time, how much is read versus pruned or shuffled, what spills, and which tasks run long after the rest.

Shuffle, skew and adaptive query execution

Adaptive query execution (AQE) is on by default. It coalesces small shuffle partitions, switches sort-merge joins to broadcast joins at runtime (on Databricks, below spark.databricks.adaptive.autoBroadcastJoinThreshold, 30 MB by default) and splits skewed partitions in sort-merge and shuffle hash joins. On classic compute, setting spark.sql.shuffle.partitions to auto enables auto-optimized shuffle, usually better than a fixed constant. AQE’s skew handling targets joins, so a hot key in an aggregation or window function still lands on one task; a salted two-stage aggregation spreads it, as in the example below. On serverless most Spark configurations cannot be set, which leaves the query and the data layout as the levers.

Joins, partitions and statistics

The optimizer chooses broadcast joins from table statistics, so stale statistics often explain a large table shuffled to join a small one. Predictive optimization runs ANALYZE on managed tables; elsewhere, compute statistics after large loads. Prefer a verified broadcast() hint to a higher job-wide threshold: broadcast data is collected on the driver and copied to every executor, and a safe threshold today becomes an out-of-memory error once the dimension grows.

Partition settings rarely need tuning beyond the defaults (128 MB read partitions and AQE coalescing). The usual culprits are inherited repartition() calls, coalesce(1) on large outputs and tiny input files. Fix small files at the table, not in every reader.

Python UDFs and caching

A row-at-a-time Python UDF serializes each row out of the JVM and back, hides its logic from the optimizer and cannot run in Photon. Most such UDFs reimplement string, date or JSON functions that Spark already has. Where Python is genuinely needed, a pandas UDF exchanges data in Arrow batches at much lower cost.

from pyspark.sql import functions as F

orders = spark.read.table("main.sales.orders")
stores = spark.read.table("main.sales.stores")  # a few thousand rows

# 1. Replace a row-at-a-time Python UDF with built-in functions.
#    Before:
#      @F.udf("string")
#      def email_domain(email):
#          return email.split("@")[-1].lower() if email else None
orders = orders.withColumn(
    "email_domain", F.lower(F.substring_index("customer_email", "@", -1))
)

# 2. Aggregate before joining, and broadcast the small side explicitly
#    when its statistics are missing or stale.
daily_revenue = (
    orders.groupBy("store_id", F.to_date("order_ts").alias("order_date"))
    .agg(F.sum("amount").alias("revenue"))
    .join(F.broadcast(stores.select("store_id", "region")), "store_id")
)

# 3. Spread a hot key across tasks with a two-stage (salted) aggregation.
#    Valid for decomposable aggregates such as sum and count.
SALT_BUCKETS = 16
lifetime_value = (
    orders.withColumn("salt", (F.rand() * SALT_BUCKETS).cast("int"))
    .groupBy("customer_id", "salt")
    .agg(F.sum("amount").alias("partial"))
    .groupBy("customer_id")
    .agg(F.sum("partial").alias("lifetime_value"))
)

df.cache() is often a cost rather than an optimization: caching a DataFrame used once adds a materialization step and memory pressure. On classic compute the disk cache already keeps local copies of Parquet and Delta files on supported instance types, and Databricks recommends it over the Spark cache for repeated reads. Spark-cached data is lost when autoscaling removes a worker, and serverless does not support CACHE TABLE. Cache only what is expensive to compute and reused within the job, then unpersist() it.

Keep table layout healthy

Layout is where storage and compute cost meet: a well-laid-out table is read with fewer files and better data skipping, by every consumer, every time.

File sizing

Small files mean more tasks, more metadata and poorer compression. Optimized writes and auto compaction (delta.autoOptimize.optimizeWrite and delta.autoOptimize.autoCompact) reduce them at write time and are always on for MERGE, UPDATE and DELETE. Unity Catalog managed tables tune their target file size automatically; for external tables, delta.targetFileSize sets an explicit target.

Liquid clustering versus partitioning and Z-ordering

Databricks recommends liquid clustering for all new tables. Up to four clustering keys, chosen from the most-filtered columns, can be changed without rewriting the table, and cannot be combined with partitioning or ZORDER. On managed tables with predictive optimization, CLUSTER BY AUTO lets the platform choose keys from observed queries. Recent runtimes can also convert a partitioned table in place with ALTER TABLE ... REPLACE PARTITIONED BY WITH CLUSTER BY; check the runtime requirement first.

-- New table: clustering keys instead of partitions.
CREATE TABLE main.sales.order_events (
  event_id  STRING,
  order_id  STRING,
  store_id  INT,
  event_ts  TIMESTAMP,
  payload   STRING
)
CLUSTER BY (store_id, event_ts);

-- Changing keys is a metadata change. Existing data is reclustered only when
-- forced, and on a large table that can take hours: schedule it.
ALTER TABLE main.sales.order_events CLUSTER BY (order_id, event_ts);
OPTIMIZE main.sales.order_events FULL;

-- Managed table with predictive optimization: let the platform choose keys.
ALTER TABLE main.sales.orders CLUSTER BY AUTO;

Predictive optimization and VACUUM

Predictive optimization runs OPTIMIZE, VACUUM and ANALYZE on Unity Catalog managed tables and is enabled by default for eligible workspaces (check the current documentation for your account). It is billed under a serverless jobs SKU, appears as PREDICTIVE_OPTIMIZATION in the usage table and logs each operation in system.storage.predictive_optimization_operations_history. It replaces maintenance you would otherwise schedule; external tables still need that scheduled maintenance.

VACUUM deletes unreferenced files older than delta.deletedFileRetentionDuration, seven days by default. A shorter retention saves storage but shortens time travel, and can delete files that a long-running write still needs. Keep the safety check that blocks short retention switched on.

Control streaming cost

A streaming query with no trigger runs as a processing-time trigger with a zero interval: it polls continuously, keeps compute up around the clock, and makes cloud storage API calls even when nothing arrives. Match the trigger to the freshness consumers need:

• processingTime with an interval bounds how often micro-batches run on compute that stays up.
• availableNow processes what is available and stops, turning a stream into a schedulable incremental batch job. It is the trigger serverless notebooks and jobs support.
• Continuous processing is experimental in open source Spark and not supported by Databricks. For sub-second latency, check the availability of Databricks’ real-time mode.

State is the other streaming cost. Aggregations, dropDuplicates and stream-stream joins keep state, and with it memory, checkpoint storage and batch duration, that grows without bound unless a watermark limits it. RocksDB is the default state store from Databricks Runtime 17.3 LTS, with changelog checkpointing on by default; enable both on older runtimes. The number of shuffle partitions for a stateful query is fixed when its checkpoint is created, so set it before the first run.

# Hourly incremental deduplication instead of an always-on stream.
# Schedule this as a job; each run processes the new data and stops.
CHECKPOINT = "/Volumes/main/sales/checkpoints/order_events_dedup"

(
    spark.readStream.table("main.sales.order_events")
    .withWatermark("event_ts", "2 hours")
    .dropDuplicatesWithinWatermark(["event_id"])
    .writeStream
    .option("checkpointLocation", CHECKPOINT)
    .trigger(availableNow=True)
    .toTable("main.sales.order_events_dedup")
)

Govern cost so the savings stay

Compute policies as guardrails

Compute policies (formerly cluster policies) constrain classic compute with fixed values, ranges, allowlists and patterns, and with the virtual dbus_per_hour attribute that caps a cluster’s hourly DBU rate. Manage them in Terraform so changes are reviewed like code:

variable "team" {
  type        = string
  description = "Owning team, written as a tag on every cluster created under the policy"
}

resource "databricks_cluster_policy" "job_compute" {
  name        = "job-compute-${var.team}"
  description = "Job compute only: LTS runtime, spot workers, bounded size, tagged"

  definition = jsonencode({
    "cluster_type"  = { type = "fixed", value = "job" }
    "spark_version" = { type = "unlimited", defaultValue = "auto:latest-lts" }
    "node_type_id" = {
      type         = "allowlist"
      values       = ["m6gd.xlarge", "m6gd.2xlarge", "r6gd.2xlarge"]
      defaultValue = "m6gd.xlarge"
    }
    "autoscale.min_workers"          = { type = "range", minValue = 1, maxValue = 2 }
    "autoscale.max_workers"          = { type = "range", maxValue = 8, defaultValue = 4 }
    "dbus_per_hour"                  = { type = "range", maxValue = 40 }
    "aws_attributes.availability"    = { type = "fixed", value = "SPOT_WITH_FALLBACK" }
    "aws_attributes.first_on_demand" = { type = "fixed", value = 1 }
    "custom_tags.team"               = { type = "fixed", value = var.team }
    "custom_tags.cost_center"        = { type = "regex", pattern = "^cc-[0-9]{4}$" }
  })
}

resource "databricks_permissions" "job_compute_use" {
  cluster_policy_id = databricks_cluster_policy.job_compute.id

  access_control {
    group_name       = "${var.team}-engineers"
    permission_level = "CAN_USE"
  }
}

The policy allows job compute only, defaults to an LTS runtime, bounds node types, workers and hourly DBUs, keeps the driver on demand, fixes the team tag and validates cost_center. The values are illustrative; set yours from measured utilization. Policies govern classic compute only; serverless relies on usage policies, budgets and review.

Budgets and alerts

Account budgets track spend by workspace and tag and email owners at thresholds. They are useful, and they are not a control.

A review cadence

• Weekly: new untagged spend, the largest day-over-day movers, and retried runs, which pay for compute again.
• Monthly: top workloads by cost per run, warehouse queueing, policy exceptions and the tuning backlog, each with a named owner.
• Quarterly: re-baseline after runtime upgrades, revisit serverless and classic choices, and retire unused jobs, warehouses and dashboards.

When migrating legacy ETL, make cost per run an acceptance criterion; the ETL migration article covers validation. The ByteLoom Lakehouse Health Check, an accelerator currently in development, is being designed as a structured assessment that covers compute efficiency alongside configuration, governance and reliability. For hands-on help, see Databricks and lakehouse engineering.

Decision summary

• One priced usage view, summing all record types, behind every cost report.
• A small, enforced tag taxonomy: compute policies for classic, usage policies for serverless.
• No scheduled work on all-purpose compute.
• Serverless or classic chosen per workload on measured cost per run, including the cloud bill.
• Warehouses sized for spill, scaled for queueing, with auto-stop set deliberately.
• Spot workers with an on-demand driver for idempotent batch; Photon tested per job.
• Query profile first; then skew, stale statistics, Python UDFs and needless caching.
• Liquid clustering for new tables, predictive optimization for managed ones.
• Streaming triggers matched to freshness, state bounded by watermarks.
• Budgets as a backstop behind a regular review with named owners.

References

• Billing system table reference
• Monitor costs using system tables
• Attribute usage with serverless usage policies
• Compute policy reference
• Photon: workloads, limitations and billing
• Adaptive query execution on Databricks
• Use liquid clustering for tables
• Predictive optimization for Unity Catalog managed tables
• Configure Structured Streaming trigger intervals
• Terraform resource: databricks_cluster_policy