The Databricks Cost Field Guide

This is your baseline. Every later finding is a share of this number. Usage is joined to the list price that was valid at the moment the usage happened, so price changes inside the window are handled correctly.

-- Q01  90-day list cost by billing product and SKU  (read-only)
-- List price, before contract discounts. Uses pricing.effective_list.default,
-- falling back to pricing.default if effective_list is empty.
WITH priced AS (
  SELECT
    u.billing_origin_product,
    u.sku_name,
    u.usage_unit,
    p.currency_code,
    u.usage_quantity,
    u.usage_quantity
      * COALESCE(p.pricing.effective_list.default, p.pricing.default) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
)
SELECT
  billing_origin_product,
  sku_name,
  usage_unit,
  currency_code,
  ROUND(SUM(usage_quantity), 2)                                AS usage_quantity_90d,
  ROUND(SUM(list_cost), 2)                                     AS list_cost_90d,
  ROUND(100 * SUM(list_cost) / SUM(SUM(list_cost)) OVER (), 1) AS pct_of_total
FROM priced
GROUP BY ALL
HAVING SUM(usage_quantity) <> 0
ORDER BY list_cost_90d DESC;

Spend concentrates. In most estates a handful of jobs carry a large share of job cost, and each one has a person who can change it. system.lakeflow.jobs keeps full history (SCD2), so the query takes the latest row per job for its current name and owner.

-- Q02  Top 10 jobs by list cost, last 90 days  (read-only)
WITH 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
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.usage_metadata.job_id IS NOT NULL
  GROUP BY ALL
),
latest_jobs AS (           -- jobs is SCD2: keep the most recent row per job
  SELECT workspace_id, job_id, name, run_as_user_name, creator_user_name, tags, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
)
SELECT
  c.workspace_id,
  c.job_id,
  j.name                                    AS job_name,
  j.run_as_user_name,
  j.creator_user_name,
  j.tags['owner']                           AS owner_tag,
  j.delete_time IS NOT NULL                 AS job_deleted,
  c.runs,
  ROUND(c.list_cost, 2)                     AS list_cost_90d,
  ROUND(c.list_cost / NULLIF(c.runs, 0), 2) AS list_cost_per_run
FROM job_cost c
LEFT JOIN latest_jobs j USING (workspace_id, job_id)
ORDER BY c.list_cost DESC
LIMIT 10;

Only jobs on jobs compute and serverless carry a job_id in billing. Jobs that run on all-purpose clusters are billed to the cluster, not the job, so they are invisible here. Q11 finds them.

Billing rows with a job_run_id can be traced to one run of one job, with a run-as identity and a schedule. Everything else needs tags or detective work. This query gives the traced share in money and percent, overall and per product.

-- Q03  Share of 90-day list cost traceable to a job run  (read-only)
WITH priced AS (
  SELECT
    u.billing_origin_product,
    u.usage_metadata.job_run_id AS job_run_id,
    u.usage_quantity
      * COALESCE(p.pricing.effective_list.default, p.pricing.default) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
)
SELECT
  COALESCE(billing_origin_product, 'ALL PRODUCTS (TOTAL)') AS billing_origin_product,
  GROUPING(billing_origin_product)                         AS is_total,
  ROUND(SUM(list_cost), 2)                                 AS list_cost_90d,
  ROUND(SUM(CASE WHEN job_run_id IS NOT NULL THEN list_cost ELSE 0 END), 2) AS traced_to_job_run,
  ROUND(SUM(CASE WHEN job_run_id IS NULL     THEN list_cost ELSE 0 END), 2) AS not_traced,
  ROUND(100 * SUM(CASE WHEN job_run_id IS NOT NULL THEN list_cost ELSE 0 END)
            / NULLIF(SUM(list_cost), 0), 1)                AS traced_pct
FROM priced
GROUP BY ROLLUP (billing_origin_product)
ORDER BY is_total DESC, list_cost_90d DESC;

system.compute.clusters is SCD2, so the query takes the latest row per cluster, then checks its tags map. Tag keys are normalised (case, dashes and underscores ignored), so Cost-Centre, cost_center and CostCentre all count. Edit the key lists to match your tagging policy.

-- Q04  Live clusters missing owner / cost-centre / environment tags  (read-only)
WITH latest_clusters AS (  -- clusters is SCD2: keep the most recent row per cluster
  SELECT *
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
),
cluster_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.cluster_id AS cluster_id,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.usage_metadata.cluster_id IS NOT NULL
  GROUP BY ALL
),
tag_check AS (
  SELECT
    c.workspace_id, c.cluster_id, c.cluster_name, c.owned_by, c.cluster_source,
    transform(COALESCE(map_keys(c.tags), array()),
              k -> lower(regexp_replace(k, '[^A-Za-z]', ''))) AS keys_norm
  FROM latest_clusters c
  WHERE c.delete_time IS NULL
    AND c.cluster_source IN ('UI', 'API')   -- remove to include job clusters
)
SELECT
  t.workspace_id,
  t.cluster_id,
  t.cluster_name,
  t.owned_by,
  t.cluster_source,
  NOT array_contains(t.keys_norm, 'owner')                            AS missing_owner,
  NOT arrays_overlap(t.keys_norm, array('costcentre', 'costcenter'))  AS missing_cost_centre,
  NOT arrays_overlap(t.keys_norm, array('environment', 'env'))        AS missing_environment,
  ROUND(COALESCE(cc.list_cost, 0), 2)                                 AS list_cost_90d
FROM tag_check t
LEFT JOIN cluster_cost cc USING (workspace_id, cluster_id)
WHERE NOT array_contains(t.keys_norm, 'owner')
   OR NOT arrays_overlap(t.keys_norm, array('costcentre', 'costcenter'))
   OR NOT arrays_overlap(t.keys_norm, array('environment', 'env'))
ORDER BY list_cost_90d DESC;

A quiet check that protects every other number. If usage has no price row for its SKU, cloud, unit and time, the joins in Q01 to Q04 drop it and your baseline is understated. This query finds what would fall through.

-- Q05  Usage with no matching list price, last 90 days  (read-only)
SELECT
  u.billing_origin_product,
  u.sku_name,
  u.cloud,
  u.usage_unit,
  COUNT(*)                        AS usage_rows,
  ROUND(SUM(u.usage_quantity), 2) AS usage_quantity_90d,
  MIN(u.usage_date)               AS first_seen,
  MAX(u.usage_date)               AS last_seen
FROM system.billing.usage u
LEFT JOIN system.billing.list_prices p
  ON  u.sku_name   = p.sku_name
  AND u.cloud      = p.cloud
  AND u.usage_unit = p.usage_unit
  AND u.usage_start_time >= p.price_start_time
  AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
  AND p.sku_name IS NULL
GROUP BY ALL
ORDER BY usage_rows DESC;

Run cost is summed from billing by job_run_id, then labelled with the run's final result_state from job_run_timeline (values: SUCCEEDED, FAILED, SKIPPED, CANCELLED, TIMED_OUT, ERROR, BLOCKED). Part A is the summary; part B ranks the jobs.

-- Q06 part A  Job run cost by final outcome, last 90 days  (read-only)
WITH run_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.job_id     AS job_id,
    u.usage_metadata.job_run_id AS run_id,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.usage_metadata.job_run_id IS NOT NULL
  GROUP BY ALL
),
run_outcome AS (           -- result_state is only set on the row that ends a run
  SELECT workspace_id, job_id, run_id, result_state, termination_code
  FROM system.lakeflow.job_run_timeline
  WHERE result_state IS NOT NULL
    AND period_end_time >= current_date() - INTERVAL 97 DAYS
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id, run_id
                             ORDER BY period_end_time DESC) = 1
)
SELECT
  o.result_state,
  COUNT(*)                                                       AS runs,
  ROUND(SUM(c.list_cost), 2)                                     AS list_cost_90d,
  ROUND(100 * SUM(c.list_cost) / SUM(SUM(c.list_cost)) OVER (), 1) AS pct_of_job_run_cost
FROM run_cost c
JOIN run_outcome o USING (workspace_id, job_id, run_id)
GROUP BY ALL
ORDER BY list_cost_90d DESC;

-- Q06 part B  Jobs ranked by failed / errored / timed-out run cost  (read-only)
WITH run_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.job_id     AS job_id,
    u.usage_metadata.job_run_id AS run_id,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.usage_metadata.job_run_id IS NOT NULL
  GROUP BY ALL
),
run_outcome AS (
  SELECT workspace_id, job_id, run_id, result_state, termination_code
  FROM system.lakeflow.job_run_timeline
  WHERE result_state IS NOT NULL
    AND period_end_time >= current_date() - INTERVAL 97 DAYS
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id, run_id
                             ORDER BY period_end_time DESC) = 1
),
latest_jobs AS (
  SELECT workspace_id, job_id, name, run_as_user_name
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
)
SELECT
  c.workspace_id,
  c.job_id,
  j.name                                         AS job_name,
  j.run_as_user_name,
  COUNT(*)                                       AS failed_runs,
  ROUND(SUM(c.list_cost), 2)                     AS failed_run_list_cost_90d,
  concat_ws(', ', collect_set(o.termination_code)) AS termination_codes
FROM run_cost c
JOIN run_outcome o USING (workspace_id, job_id, run_id)
LEFT JOIN latest_jobs j USING (workspace_id, job_id)
WHERE o.result_state IN ('FAILED', 'ERROR', 'TIMED_OUT')
GROUP BY ALL
ORDER BY failed_run_list_cost_90d DESC
LIMIT 20;

When a run is repaired, job_run_timeline records another row with a result state for the same run_id, so terminal rows minus one equals repair attempts (the same logic Databricks uses in its own repair query). Part B counts task attempts beyond one per task per run, which captures automatic task retries.

-- Q07 part A  Jobs with the most repaired runs, last 90 days  (read-only)
WITH run_attempts AS (
  SELECT
    workspace_id, job_id, run_id,
    COUNT(*)                              AS terminal_rows,
    MAX_BY(result_state, period_end_time) AS final_state
  FROM system.lakeflow.job_run_timeline
  WHERE result_state IS NOT NULL
    AND period_end_time >= current_date() - INTERVAL 90 DAYS
  GROUP BY ALL
),
latest_jobs AS (
  SELECT workspace_id, job_id, name, run_as_user_name
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
)
SELECT
  r.workspace_id,
  r.job_id,
  j.name                                                    AS job_name,
  j.run_as_user_name,
  COUNT(*)                                                  AS runs,
  SUM(CASE WHEN r.terminal_rows > 1 THEN 1 ELSE 0 END)      AS repaired_runs,
  SUM(r.terminal_rows - 1)                                  AS repair_attempts,
  ROUND(100 * SUM(CASE WHEN r.terminal_rows > 1 THEN 1 ELSE 0 END) / COUNT(*), 1)
                                                            AS repaired_pct,
  SUM(CASE WHEN r.final_state <> 'SUCCEEDED' THEN 1 ELSE 0 END) AS runs_not_succeeded
FROM run_attempts r
LEFT JOIN latest_jobs j USING (workspace_id, job_id)
GROUP BY ALL
HAVING SUM(r.terminal_rows - 1) > 0
ORDER BY repair_attempts DESC
LIMIT 20;

-- Q07 part B  Tasks that needed more than one attempt per run, last 90 days  (read-only)
SELECT
  workspace_id,
  job_id,
  task_key,
  COUNT(DISTINCT job_run_id)                          AS job_runs,
  COUNT(DISTINCT run_id)                              AS task_attempts,
  COUNT(DISTINCT run_id) - COUNT(DISTINCT job_run_id) AS extra_attempts
FROM system.lakeflow.job_task_run_timeline
WHERE result_state IS NOT NULL
  AND period_end_time >= current_date() - INTERVAL 90 DAYS
GROUP BY ALL
HAVING COUNT(DISTINCT run_id) - COUNT(DISTINCT job_run_id) > 0
ORDER BY extra_attempts DESC
LIMIT 25;

Both jobs and job_tasks carry a timeout_seconds column, where 0 or empty means no limit. A hung job with no timeout bills until someone notices. The query lists live jobs where the job, or any of its tasks, has no timeout, with recent run count and longest run.

-- Q08  Live jobs with no job-level or task-level timeout  (read-only)
WITH latest_jobs AS (
  SELECT *
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
latest_tasks AS (  -- job_tasks is SCD2 too
  SELECT *
  FROM system.lakeflow.job_tasks
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id, task_key ORDER BY change_time DESC) = 1
),
task_timeouts AS (
  SELECT
    workspace_id, job_id,
    COUNT(*)                                                     AS tasks,
    SUM(CASE WHEN COALESCE(timeout_seconds, 0) = 0 THEN 1 ELSE 0 END) AS tasks_without_timeout
  FROM latest_tasks
  WHERE delete_time IS NULL
  GROUP BY ALL
),
recent_runs AS (
  SELECT
    workspace_id, job_id,
    COUNT(DISTINCT run_id)                                         AS runs_90d,
    MAX(CASE WHEN result_state IS NOT NULL THEN run_duration_seconds END) AS longest_run_seconds
  FROM system.lakeflow.job_run_timeline
  WHERE period_end_time >= current_date() - INTERVAL 90 DAYS
  GROUP BY ALL
)
SELECT
  j.workspace_id,
  j.job_id,
  j.name                                      AS job_name,
  j.run_as_user_name,
  j.paused,
  COALESCE(j.timeout_seconds, 0)              AS job_timeout_seconds,
  t.tasks,
  t.tasks_without_timeout,
  COALESCE(r.runs_90d, 0)                     AS runs_90d,
  ROUND(r.longest_run_seconds / 3600.0, 2)    AS longest_run_hours
FROM latest_jobs j
LEFT JOIN task_timeouts t USING (workspace_id, job_id)
LEFT JOIN recent_runs  r USING (workspace_id, job_id)
WHERE j.delete_time IS NULL
  AND (COALESCE(j.timeout_seconds, 0) = 0 OR t.tasks_without_timeout > 0)
ORDER BY runs_90d DESC, longest_run_hours DESC;

node_timeline records one row per node per minute. CPU busy is cpu_user_percent + cpu_system_percent. Drivers are excluded unless the cluster is single-node, because an idle driver would drag every average down. The window is 30 days to keep the scan light; widen it to 90 if you like.

-- Q09  CPU busy % per cluster (avg and p95), last 30 days  (read-only)
WITH nodes AS (
  SELECT
    workspace_id, cluster_id, driver, start_time, mem_used_percent,
    cpu_user_percent + cpu_system_percent AS cpu_busy_pct,
    MAX(CASE WHEN driver THEN 0 ELSE 1 END)
      OVER (PARTITION BY workspace_id, cluster_id) AS has_workers
  FROM system.compute.node_timeline
  WHERE start_time >= current_date() - INTERVAL 30 DAYS
),
per_cluster AS (
  SELECT
    workspace_id, cluster_id,
    COUNT(DISTINCT date_trunc('MINUTE', start_time)) AS active_minutes,
    AVG(cpu_busy_pct)                                AS avg_cpu_busy_pct,
    PERCENTILE_APPROX(cpu_busy_pct, 0.95)            AS p95_cpu_busy_pct,
    AVG(mem_used_percent)                            AS avg_mem_used_pct
  FROM nodes
  WHERE driver = false OR has_workers = 0   -- workers only, unless single-node
  GROUP BY ALL
),
latest_clusters AS (
  SELECT workspace_id, cluster_id, cluster_name, cluster_source, worker_node_type, owned_by
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
),
cluster_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.cluster_id AS cluster_id,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 30 DAYS
    AND u.usage_metadata.cluster_id IS NOT NULL
  GROUP BY ALL
)
SELECT
  pc.workspace_id,
  pc.cluster_id,
  c.cluster_name,
  c.cluster_source,
  c.worker_node_type,
  c.owned_by,
  ROUND(pc.active_minutes / 60.0, 1)  AS active_hours,
  ROUND(pc.avg_cpu_busy_pct, 1)       AS avg_cpu_busy_pct,
  ROUND(pc.p95_cpu_busy_pct, 1)       AS p95_cpu_busy_pct,
  ROUND(pc.avg_mem_used_pct, 1)       AS avg_mem_used_pct,
  ROUND(COALESCE(cc.list_cost, 0), 2) AS list_cost_30d,
  (pc.active_minutes >= 30
   AND pc.avg_cpu_busy_pct < 20
   AND pc.p95_cpu_busy_pct < 40)      AS under_utilised
FROM per_cluster pc
LEFT JOIN latest_clusters c  USING (workspace_id, cluster_id)
LEFT JOIN cluster_cost    cc USING (workspace_id, cluster_id)
ORDER BY under_utilised DESC, list_cost_30d DESC;

Two configuration smells in one pass over the latest cluster definitions. When min_autoscale_workers equals max_autoscale_workers, autoscaling is switched off in practice. When auto_termination_minutes is 0, an interactive cluster runs until someone stops it.

-- Q10  Live interactive clusters: autoscale min = max, or no / long auto-termination  (read-only)
WITH latest_clusters AS (
  SELECT *
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
)
SELECT
  workspace_id,
  cluster_id,
  cluster_name,
  cluster_source,
  owned_by,
  worker_node_type,
  worker_count,
  min_autoscale_workers,
  max_autoscale_workers,
  auto_termination_minutes,
  concat_ws(' | ',
    CASE WHEN min_autoscale_workers IS NOT NULL
          AND min_autoscale_workers = max_autoscale_workers
         THEN 'Autoscale min = max' END,
    CASE WHEN COALESCE(auto_termination_minutes, 0) = 0
         THEN 'No auto-termination' END,
    CASE WHEN auto_termination_minutes > 60
         THEN 'Idle window over 60 min' END
  ) AS findings
FROM latest_clusters
WHERE delete_time IS NULL
  AND cluster_source IN ('UI', 'API')
  AND (   (min_autoscale_workers IS NOT NULL AND min_autoscale_workers = max_autoscale_workers)
       OR COALESCE(auto_termination_minutes, 0) = 0
       OR auto_termination_minutes > 60)
ORDER BY findings, cluster_name;

Databricks bills jobs that run on all-purpose clusters to the cluster, not the job: usage_metadata.job_id is not populated for them. So filtering billing on ALL_PURPOSE SKUs with a job id returns nothing. Instead, this query reads compute_ids on each task run, keeps the ones that point at interactive clusters (source UI or API), then prices the cluster-hours those tasks touched.

-- Q11  Jobs whose tasks ran on all-purpose clusters, last 90 days  (read-only)
WITH latest_clusters AS (
  SELECT workspace_id, cluster_id, cluster_name, cluster_source, owned_by
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
),
task_clusters AS (
  SELECT t.workspace_id, t.job_id, t.period_start_time,
         explode(t.compute_ids) AS cluster_id
  FROM system.lakeflow.job_task_run_timeline t
  WHERE t.period_start_time >= current_date() - INTERVAL 90 DAYS
),
task_hours AS (            -- each hour in which a scheduled task touched an interactive cluster
  SELECT DISTINCT
    tc.workspace_id, tc.job_id, tc.cluster_id, c.cluster_name, c.owned_by,
    date_trunc('HOUR', tc.period_start_time) AS usage_hour
  FROM task_clusters tc
  JOIN latest_clusters c USING (workspace_id, cluster_id)
  WHERE c.cluster_source IN ('UI', 'API')
),
cluster_hour_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.cluster_id               AS cluster_id,
    date_trunc('HOUR', u.usage_start_time)    AS usage_hour,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.sku_name LIKE '%ALL_PURPOSE%'
  GROUP BY ALL
),
latest_jobs 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
)
SELECT
  th.workspace_id,
  th.job_id,
  j.name                         AS job_name,
  th.cluster_id,
  th.cluster_name,
  th.owned_by                    AS cluster_owner,
  COUNT(*)                       AS cluster_hours_touched,
  ROUND(SUM(chc.list_cost), 2)   AS est_all_purpose_list_cost_90d
FROM task_hours th
LEFT JOIN cluster_hour_cost chc USING (workspace_id, cluster_id, usage_hour)
LEFT JOIN latest_jobs j         USING (workspace_id, job_id)
GROUP BY ALL
ORDER BY est_all_purpose_list_cost_90d DESC NULLS LAST;

For each live job: task count and dependency edges from job_tasks (depends_on_keys), a structural complexity band, and three dependency signals billing can see for you: serverless, Photon and GPU usage. The others (dbutils, Databricks SDK, Lakeflow Declarative Pipelines, ML, streaming) need a look at the code.

-- Q12  Job complexity and dependency worksheet, last 90 days  (read-only)
WITH latest_jobs AS (
  SELECT *
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
latest_tasks AS (
  SELECT *
  FROM system.lakeflow.job_tasks
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id, task_key ORDER BY change_time DESC) = 1
),
task_shape AS (
  SELECT
    workspace_id, job_id,
    COUNT(*) AS task_count,
    SUM(CASE WHEN depends_on_keys IS NULL THEN 0 ELSE size(depends_on_keys) END) AS dependency_edges,
    MAX(CASE WHEN depends_on_keys IS NULL THEN 0 ELSE size(depends_on_keys) END) AS max_fan_in
  FROM latest_tasks
  WHERE delete_time IS NULL
  GROUP BY ALL
),
gpu_nodes AS (
  SELECT DISTINCT node_type FROM system.compute.node_types WHERE gpu_count > 0
),
job_usage AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.job_id AS job_id,
    SUM(u.usage_quantity
        * COALESCE(p.pricing.effective_list.default, p.pricing.default)) AS list_cost,
    bool_or(u.product_features.is_serverless)  AS uses_serverless,
    bool_or(u.product_features.is_photon)      AS uses_photon,
    bool_or(g.node_type IS NOT NULL)           AS uses_gpu
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.sku_name   = p.sku_name
    AND u.cloud      = p.cloud
    AND u.usage_unit = p.usage_unit
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  LEFT JOIN gpu_nodes g ON g.node_type = u.usage_metadata.node_type
  WHERE u.usage_date >= current_date() - INTERVAL 90 DAYS
    AND u.usage_metadata.job_id IS NOT NULL
  GROUP BY ALL
)
SELECT
  j.workspace_id,
  j.job_id,
  j.name                                   AS job_name,
  j.run_as_user_name,
  j.trigger_type,
  s.task_count,
  s.dependency_edges,
  s.max_fan_in,
  CASE WHEN s.task_count >= 10 OR s.dependency_edges >= 12 THEN 'High'
       WHEN s.task_count >= 4  OR s.dependency_edges >= 4  THEN 'Medium'
       ELSE 'Low' END                      AS structural_complexity,
  COALESCE(ju.uses_serverless, false)      AS uses_serverless,
  COALESCE(ju.uses_photon, false)          AS uses_photon,
  COALESCE(ju.uses_gpu, false)             AS uses_gpu,
  ROUND(COALESCE(ju.list_cost, 0), 2)      AS list_cost_90d
FROM latest_jobs j
JOIN task_shape s USING (workspace_id, job_id)
LEFT JOIN job_usage ju USING (workspace_id, job_id)
WHERE j.delete_time IS NULL
ORDER BY list_cost_90d DESC;

Pick a complexity band and tick the Databricks dependencies you found. Lakeflow Declarative Pipelines (formerly DLT), ML and streaming weigh 2 points each because they are the hardest to replicate elsewhere; the rest weigh 1. Low dependency is 0 to 1 points, Medium 2 to 3, High 4 or more. Scores stay in this browser.

A starting heuristic, not a verdict. Defer anything whose 90-day cost would not pay for the effort of moving it, whatever the matrix says.

Generic SQL templates. Fill in the fields and every template below rewrites itself. Use three-part names. If the target lives outside Databricks, land its output in a comparison schema first, or run the same SQL on the target engine and compare results side by side.

Across engines, cast floating-point columns to a fixed DECIMAL and normalise timestamps to UTC before hashing, or identical data will hash differently. Record any accepted tolerance in the change record.