How to calculate cost of each table for the specific Databricks Run ID

[Dolly0503]

Hi Databricks Community,

I need help calculating table-level cost for each specific Job Run ID in Databricks.

I have multiple pipelines/jobs, and the same table can run multiple times with different Run IDs.

For example:
Job A
│
├── Run ID 1001
│ ├── Table A → $5
│ └── Table B → $3
│
├── Run ID 1002
│ ├── Table A → $7
│ └── Table C → $4
│
└── Run ID 1003
└── Table A → $6

I need the o/p to be 

Job ID Run ID Table Table Cost

Job A	1001	Table A	$5
Job A	1001	Table B	$3
Job A	1002	Table A	$7
Job A	1002	Table C	$4
Job A	1003	Table A	$6

The key requirement is:

If a table was executed in a particular Run ID, I want to see the cost of that table specifically for that Run ID.

I do not want:

Table A = $18

without knowing which Run IDs contributed to that $18.

I want:

Table A
  Run 1001 → $5
  Run 1002 → $7
  Run 1003 → $6

I am currently exploring:

• system.billing.usage
• system.billing.list_prices
• system.lakeflow.job_task_run_timeline
• system.lakeflow.jobs

I would like to know the recommended Databricks approach to derive:

Job ID
→ Run ID
→ Task
→ Table
→ Table Cost

The solution should also ensure that the sum of table-level costs for a Run ID reconciles with the actual cost of that Run ID, without double counting.

What is the best Databricks-native way to achieve this table-level cost attribution by Run ID? and also i need end to end query

[bhawana-pandey]

Databricks has no native table-level cost. You proportionally allocate run cost based on task duration.

Table Cost = Run Cost × (Task Duration / Total Run Duration) ÷ Tables per Task

Tables Needed

Table Purpose

system.billing.usage	DBU consumption per run
system.billing.list_prices	DBU → $ rate
system.lakeflow.job_task_run_timeline	Task duration per run
system.access.audit	Which tables a task touched

The Query (4 CTEs)

WITH run_costs AS (
SELECT custom_tags['jobId'] AS job_id, custom_tags['runId'] AS run_id,
SUM(u.usage_quantity * lp.pricing.default) AS run_cost_usd
FROM system.billing.usage u
JOIN system.billing.list_prices lp ON u.sku_name = lp.sku_name
WHERE billing_origin_product = 'JOBS'
GROUP BY 1, 2
),
task_weights AS (
SELECT job_id, run_id, task_key,
(unix_timestamp(result_time) - unix_timestamp(start_time)) * 1.0 /
NULLIF(SUM(unix_timestamp(result_time) - unix_timestamp(start_time))
OVER (PARTITION BY job_id, run_id), 0) AS task_weight
FROM system.lakeflow.job_task_run_timeline
),
task_tables AS (
SELECT DISTINCT request_params['jobId'] AS job_id, request_params['runId'] AS run_id,
request_params['taskKey'] AS task_key,
CONCAT_WS('.', request_params['catalogName'], request_params['schemaName'],
request_params['tableName']) AS table_name
FROM system.access.audit
WHERE service_name = 'unityCatalog' AND request_params['jobId'] IS NOT NULL
),
table_counts AS (
SELECT job_id, run_id, task_key, COUNT(*) AS table_cnt FROM task_tables GROUP BY 1,2,3
)
SELECT j.name AS job_name, tw.job_id, tw.run_id, tw.task_key, tt.table_name,
ROUND(rc.run_cost_usd * tw.task_weight / tc.table_cnt, 4) AS table_cost_usd
FROM task_weights tw
JOIN run_costs rc ON tw.job_id = rc.job_id AND tw.run_id = rc.run_id
JOIN task_tables tt ON tw.job_id = tt.job_id AND tw.run_id = tt.run_id AND tw.task_key = tt.task_key
JOIN table_counts tc ON tw.job_id = tc.job_id AND tw.run_id = tc.run_id AND tw.task_key = tc.task_key
LEFT JOIN system.lakeflow.jobs j ON tw.job_id = j.job_id
ORDER BY tw.job_id, tw.run_id, tt.table_name;

[Dolly0503]

Hi Bhawana,

I tried to implement the code as shown below, but how can I validate whether it is returning the correct results? Also, we are planning to enable the Genie workspace — if someone provides a table name, it should return the table name, run ID, and average cost of that runs (
select job_id, day, sum(cost_consumed) as cost_val
from schemaname.tablename
group by job_id, day
),

-- job cte

base_data as (
select *,
setup_duration_seconds + execution_duration_seconds as total_run_duration,
to_date(period_start_time) as execution_date
from system.lakeflow.job_task_run_timeline
where workspace_id IN ('123456', '107893')
),

job_names as (
select job_id, name as job_name
from system.lakeflow.jobs
),

data as (
select
*,
avg(execution_duration_seconds) over (
partition by job_id, to_date(period_start_time), task_key
) as avg_task_exec_for_day,

avg(sum(execution_duration_seconds)) over (
partition by job_id, to_date(period_start_time), parent_run_id
) over (partition by job_id, to_date(period_start_time))
as avg_exec_for_day
from base_data
)

select distinct
d.job_id,
j.job_name,
d.execution_date,
d.task_key,
split_part(d.task_key, '-', 1) as ctlg,
split_part(d.task_key, '-', 2) as db_name,
split_part(d.task_key, '-', 3) as tbl_name,
d.avg_task_exec_for_day,
d.avg_exec_for_day,
round(d.avg_task_exec_for_day / d.avg_exec_for_day, 3) as task_lvls_weightage,
round(c.cost_val, 5) as job_lvl_cost_in_usd,
round(task_lvls_weightage * c.cost_val, 5) as task_lvl_cost_in_usd
from data d
left join cost_agg c on c.job_id = d.job_id and c.day = d.execution_date
left join job_names j on d.job_id = j.job_id
where d.job_id = 155301936099512
order by execution_date desc, task_lvls_weightage desc

[PuchninSergei]

@Dolly0503,

yes the idea is the same what @bhawana-pandey said.

if you need more accurate - use " (DESCRIBE HISTORY demo.sales.order_margin)" to get table size.
 
You can try to
1. from system.billing.usage and system.billing.list_prices price every JOBS usage -> take workspace_id + job_id + job_run_id
2. from system.lakeflow.job_task_run_timeline get active seconds per TASK within each run
3. Allocate run cost (#1) to tasks (#2) by duration share to cost per task
4. from system.access.table_lineage link tables to runs (entity_run_id is job_run_id)
5. split cost for each Run ID (#3) equally across the tables of that run (#4)

but that splits cost equally across the tables.
not sure how split more accurate.

As a start idea - use history information to find how big was the specific change:

SELECT
version,
timestamp,
operation,
operationMetrics['numOutputRows'] AS num_output_rows,
operationMetrics['numOutputBytes'] AS num_output_bytes,
operationMetrics['numFiles'] AS num_files
FROM (DESCRIBE HISTORY demo.sales.order_margin)
WHERE job.runId IS NOT NULL
AND operation IN ('WRITE','MERGE','CREATE TABLE AS SELECT','INSERT')

[Satyasai]

To calculate table-level cost for each specific Job Run ID in Databricks, you must join system billing logs with operational metadata.

Databricks billing metrics (system.billing.usage) track cost down to the Task Run ID level, but Databricks does not track table-level cost natively inside billing logs, because a single Spark task or write operation might process multiple tables.

To achieve 100% reconciliation (where the sum of table costs for a Run ID equals the actual Run ID billing cost without double counting or under counting), the recommended approach is to allocate the Run ID's billed cost proportionally across the tables written during that specific run based on the bytes written or rows inserted.

Core Logic & Architecture

1.  Billing Base: Join system.billing.usage with system.billing.list_prices to get the exact cost per job_run_id and task_id (custom_tags.ResourceClass or usage_metadata.job_run_id).

Query –

SELECT

        u.usage_metadata.job_id AS job_id,

        u.usage_metadata.job_run_id AS run_id,

        u.usage_metadata.task_id AS task_id,

        u.account_id,

        u.cloud,

        SUM(u.usage_quantity * COALESCE(p.pricing.default, 0)) AS task_run_cost

    FROM system.billing.usage u

    LEFT JOIN system.billing.list_prices p

        ON u.sku_name = p.sku_name

        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_metadata.job_run_id IS NOT NULL

    GROUP BY ALL

2.  Table Operations (Delta Query Log): Query Delta Lake transaction history via system.access.audit or Delta table_history to identify which tables were modified by each job_run_id and the volume written (numOutputBytes / numOutputRows).

SELECT

        request_params.job_id AS job_id,

        request_params.job_run_id AS run_id,

        request_params.task_id AS task_id,

        CONCAT(request_params.catalog_name, '.', request_params.schema_name, '.', request_params.table_name) AS table_name,

        -- Fallback to 1 if output bytes are missing/zero to allow equal split

        COALESCE(CAST(response.result.numOutputBytes AS DOUBLE), 1.0) AS bytes_written

    FROM system.access.audit

    WHERE service_name = 'unityCatalog'

      AND action_name IN ('createTable', 'modifyTable', 'writeTable', 'deltaCommit')

      AND request_params.job_run_id IS NOT NULL

3.  Proportional Allocation: Distribute the task/run cost to each table using:

SELECT

        job_id,

        run_id,

        task_id,

        table_name,

        bytes_written,

        SUM(bytes_written) OVER (PARTITION BY job_id, run_id, task_id) AS total_task_bytes

    FROM table_write_metrics

Final Query –

WITH job_run_costs AS (

    -- Step 1: Calculate total cost per Job ID, Run ID, and Task ID from Billing Logs

    SELECT

        u.usage_metadata.job_id AS job_id,

        u.usage_metadata.job_run_id AS run_id,

        u.usage_metadata.task_id AS task_id,

        u.account_id,

        u.cloud,

        SUM(u.usage_quantity * COALESCE(p.pricing.default, 0)) AS task_run_cost

    FROM system.billing.usage u

    LEFT JOIN system.billing.list_prices p

        ON u.sku_name = p.sku_name

        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_metadata.job_run_id IS NOT NULL

    GROUP BY ALL

),

table_write_metrics AS (

    -- Step 2: Extract table write activity per Job Run ID from System Audit Logs

    SELECT

        request_params.job_id AS job_id,

        request_params.job_run_id AS run_id,

        request_params.task_id AS task_id,

        CONCAT(request_params.catalog_name, '.', request_params.schema_name, '.', request_params.table_name) AS table_name,

        -- Fallback to 1 if output bytes are missing/zero to allow equal split

        COALESCE(CAST(response.result.numOutputBytes AS DOUBLE), 1.0) AS bytes_written

    FROM system.access.audit

    WHERE service_name = 'unityCatalog'

      AND action_name IN ('createTable', 'modifyTable', 'writeTable', 'deltaCommit')

      AND request_params.job_run_id IS NOT NULL

),

task_table_totals AS (

    -- Step 3: Compute total bytes written per Task Run ID for cost distribution

    SELECT

        job_id,

        run_id,

        task_id,

        table_name,

        bytes_written,

        SUM(bytes_written) OVER (PARTITION BY job_id, run_id, task_id) AS total_task_bytes

    FROM table_write_metrics

)

-- Step 4: Proportional Allocation of Cost to Tables per Run ID

SELECT

    c.job_id AS `Job ID`,

    c.run_id AS `Run ID`,

    c.task_id AS `Task`,

    COALESCE(t.table_name, 'Non-Table / Overhead Compute') AS `Table`,

    ROUND(

        c.task_run_cost * (COALESCE(t.bytes_written, 1.0) / COALESCE(t.total_task_bytes, 1.0)),

        2

    ) AS `Table Cost`

FROM job_run_costs c

LEFT JOIN task_table_totals t

    ON c.job_id = t.job_id

   AND c.run_id = t.run_id

   AND c.task_id = t.task_id

ORDER BY c.job_id, c.run_id, c.task_id, `Table Cost` DESC;