Backfilling Historical Cost with Idempotent Batch Jobs

Loading six months of historical database cost into a fresh ledger is a one-shot job that almost never finishes on the first attempt, so this page builds a day-chunked backfill that pages AWS Cost Explorer, upserts on a natural key, and resumes from a checkpoint table without ever double-counting.

Back to: Batch Processing for Historical Metrics

A backfill differs from an incremental pull in one decisive way: it touches a large, closed date range in a single run, so any failure — a throttled API, a dropped connection, a redeploy mid-job — lands you halfway through with no clean way to tell what already committed. The fix is to make the whole job idempotent: chunk the range by day, write each day through an ON CONFLICT DO UPDATE upsert keyed on (tenant_id, resource_id, billing_period, export_version), and record every completed day in a checkpoint table. Re-running then skips finished days and rewrites in-flight ones to the exact same values. This is the batch counterpart to the streaming work in real-time metric streaming setup, and every record it writes must first clear strict schema validation for billing data. Cost Explorer itself restates recent days for up to 72 hours, which is exactly why the upsert must overwrite rather than insert — a resumed run naturally picks up the corrected figures.

Prerequisites

Before running the backfill, confirm the following are in place.

  • AWS IAM: the job needs read-only Cost Explorer access and nothing else. Attach a policy scoped to the two Cost Explorer read actions; least-privilege scoping here is part of broader access control for cost data.

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "CostExplorerRead",
      "Effect": "Allow",
      "Action": [
        "ce:GetCostAndUsage",
        "ce:GetCostAndUsageWithResources"
      ],
      "Resource": "*"
    }
  ]
}

  • Python: 3.10 or newer (the code uses modern typing and datetime APIs).

  • Libraries: install the AWS SDK and the Postgres 3 driver.

    pip install "boto3>=1.34" "psycopg[binary]>=3.1"

  • Target schema: a fact table with a natural unique key and a companion checkpoint table.

    CREATE TABLE cost_fact (
    tenant_id       text        NOT NULL,
    resource_id     text        NOT NULL,
    billing_period  date        NOT NULL,   -- the day this cost belongs to
    export_version  int         NOT NULL,   -- bumped when the source restates
    amount          numeric(18,6) NOT NULL,
    currency        text        NOT NULL,
    loaded_at       timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (tenant_id, resource_id, billing_period, export_version)
);

CREATE TABLE backfill_checkpoint (
    job_id          text NOT NULL,
    billing_period  date NOT NULL,
    rows_written    int  NOT NULL,
    completed_at    timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (job_id, billing_period)
);

Step-by-Step Implementation

The backfill enumerates the date range one day at a time, checks the checkpoint to skip days already done, pages Cost Explorer for the remaining days, upserts each day’s rows in a single transaction with the checkpoint write, and treats throttling as a retryable pause. Build it in four steps.

Step 1 — Enumerate day chunks and read the checkpoint

Chunking by day gives the smallest safely-restartable unit: a day either commits completely or not at all. Before doing any work, load the set of days this job_id already finished so a resumed run does zero redundant API calls.

import os
from datetime import date, timedelta

import psycopg


def day_range(start: date, end: date) -> list[date]:
    """Inclusive-start, exclusive-end list of days to backfill."""
    return [start + timedelta(days=i) for i in range((end - start).days)]


def completed_days(conn: psycopg.Connection, job_id: str) -> set[date]:
    """Days already committed for this job, so a resume can skip them."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT billing_period FROM backfill_checkpoint WHERE job_id = %s",
            (job_id,),
        )
        return {row[0] for row in cur.fetchall()}


# Expected:
#   day_range(date(2026,1,1), date(2026,1,4)) -> [2026-01-01, 2026-01-02, 2026-01-03]
#   completed_days(conn, "ce-backfill-2026H1") -> {date(2026, 1, 1)}  on a resumed run

Step 2 — Page one day out of Cost Explorer

Cost Explorer returns a NextPageToken when a day’s grouped result exceeds one page, and it raises a throttling fault under load. Fetch a full day — every page — grouping by the dimensions that become your natural key, and let a throttle bubble up as a typed signal the caller retries.

import time
import logging

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger("ce_backfill")


def fetch_day(client, day: date) -> list[dict]:
    """Return every grouped cost row for a single day, following pagination."""
    start = day.isoformat()
    end = (day + timedelta(days=1)).isoformat()
    rows: list[dict] = []
    next_token: str | None = None

    while True:
        kwargs = {
            "TimePeriod": {"Start": start, "End": end},
            "Granularity": "DAILY",
            "Metrics": ["UnblendedCost"],
            "GroupBy": [
                {"Type": "TAG", "Key": "tenant_id"},
                {"Type": "DIMENSION", "Key": "RESOURCE_ID"},
            ],
        }
        if next_token:
            kwargs["NextPageToken"] = next_token
        resp = client.get_cost_and_usage_with_resources(**kwargs)

        for result in resp["ResultsByTime"]:
            for group in result["Groups"]:
                tenant = group["Keys"][0].removeprefix("tenant_id$") or "untagged"
                amount = group["Metrics"]["UnblendedCost"]
                rows.append({
                    "tenant_id": tenant,
                    "resource_id": group["Keys"][1],
                    "billing_period": day,
                    "amount": float(amount["Amount"]),
                    "currency": amount["Unit"],
                })
        next_token = resp.get("NextPageToken")
        if not next_token:
            return rows

A tag group key arrives prefixed as tenant_id$acme; removeprefix strips it, and an empty remainder means the resource carried no tenant_id tag, which we bucket as untagged rather than dropping.

Step 3 — Upsert the day and checkpoint in one transaction

This is the heart of idempotency. Every row goes through ON CONFLICT (tenant_id, resource_id, billing_period, export_version) DO UPDATE, and the checkpoint row is written in the same transaction as the facts. Either the day’s data and its checkpoint both commit, or neither does — there is no window where a day is loaded but unmarked, or marked but unloaded.

UPSERT_SQL = """
    INSERT INTO cost_fact
        (tenant_id, resource_id, billing_period, export_version, amount, currency)
    VALUES (%(tenant_id)s, %(resource_id)s, %(billing_period)s,
            %(export_version)s, %(amount)s, %(currency)s)
    ON CONFLICT (tenant_id, resource_id, billing_period, export_version)
    DO UPDATE SET
        amount   = EXCLUDED.amount,
        currency = EXCLUDED.currency,
        loaded_at = now()
"""

CHECKPOINT_SQL = """
    INSERT INTO backfill_checkpoint (job_id, billing_period, rows_written)
    VALUES (%(job_id)s, %(billing_period)s, %(rows_written)s)
    ON CONFLICT (job_id, billing_period)
    DO UPDATE SET rows_written = EXCLUDED.rows_written, completed_at = now()
"""


def commit_day(conn, job_id: str, day: date, rows: list[dict], export_version: int):
    """Atomically upsert one day's rows and its checkpoint."""
    for r in rows:
        r["export_version"] = export_version
    with conn.transaction():                      # single all-or-nothing unit
        with conn.cursor() as cur:
            cur.executemany(UPSERT_SQL, rows)
            cur.execute(CHECKPOINT_SQL, {
                "job_id": job_id,
                "billing_period": day,
                "rows_written": len(rows),
            })
    logger.info("day %s committed: %d rows", day, len(rows))

Because the upsert overwrites on the natural key, replaying a day that Cost Explorer has since restated simply writes the corrected amount. Bumping export_version is reserved for the case where you must keep both the old and restated figures side by side for audit; a plain resume leaves it fixed.

Step 4 — Drive the range with resume and throttle handling

The driver skips completed days, retries a throttled day with exponential backoff instead of failing the whole job, and leaves an unfinished day uncheckpointed so the next run retries it cleanly. The control flow is shown below.

Idempotent backfill driver flow: skip checkpointed days, fetch, retry on throttle, atomically upsert and checkpointFor each day in the range the driver checks the checkpoint. A checkpointed day is skipped. An unfinished day is fetched from Cost Explorer; a throttling exception triggers exponential backoff and a retry of the same day, while a successful fetch upserts the rows and writes the checkpoint atomically before advancing. When no days remain the backfill completes.already donenew daythrottledretryrowsnext dayNext day in rangeIncheckpoint?SkipFetch dayCost Explorer, all pagesThrottled?Backoffsleep 2ⁿ, retryUpsert + checkpointone transactionRange complete
from botocore.config import Config

MAX_RETRIES = 5
BASE_DELAY = 2.0


def run_backfill(job_id: str, start: date, end: date, export_version: int = 1) -> None:
    """Backfill [start, end) day by day, resumable and idempotent."""
    client = boto3.client(
        "ce",
        config=Config(retries={"max_attempts": 0}),  # we own retry, not botocore
    )
    with psycopg.connect(os.environ["PG_DSN"]) as conn:
        done = completed_days(conn, job_id)
        for day in day_range(start, end):
            if day in done:
                logger.info("day %s already checkpointed, skipping", day)
                continue
            for attempt in range(MAX_RETRIES):
                try:
                    rows = fetch_day(client, day)
                    commit_day(conn, job_id, day, rows, export_version)
                    break
                except ClientError as exc:
                    code = exc.response["Error"]["Code"]
                    if code in ("ThrottlingException", "LimitExceededException") \
                            and attempt < MAX_RETRIES - 1:
                        delay = BASE_DELAY * (2 ** attempt)
                        logger.warning("throttled on %s, retry in %.0fs", day, delay)
                        time.sleep(delay)
                        continue
                    raise
    logger.info("backfill %s complete", job_id)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run_backfill("ce-backfill-2026H1", date(2026, 1, 1), date(2026, 7, 1))

Expected output on a run that was interrupted after March and restarted:

day 2026-01-01 already checkpointed, skipping
...
day 2026-03-31 already checkpointed, skipping
day 2026-04-01 committed: 412 rows
day 2026-04-02 committed: 407 rows
...
backfill ce-backfill-2026H1 complete

Verification

Prove the job is idempotent before trusting it against a real ledger.

  1. Run it twice; the fact table must be byte-identical. Snapshot a checksum, re-run, and compare — a correct upsert overwrites in place and adds no rows.

    psql "$PG_DSN" -tAc \
  "SELECT md5(string_agg(tenant_id||resource_id||billing_period::text||amount::text, ',' ORDER BY 1,2,3)) FROM cost_fact"

    The digest must match before and after the second run.

  2. Assert the checkpoint covers every day. No gaps means no silently-missing day.

    done = completed_days(conn, "ce-backfill-2026H1")
assert done == set(day_range(date(2026, 1, 1), date(2026, 7, 1))), "missing days"

  3. Inspect one committed row. A healthy fact is shaped like ('acme', 'arn:aws:rds:...:db:prod-1', 2026-04-01, 1, 18.42, 'USD') — the shape the aggregation tier consumes directly.

Gotchas & Edge Cases

  • Cost Explorer restates recent days. Figures for the last ~72 hours are provisional and change as usage finalizes. Backfill closed history freely, but re-run the trailing three days on a schedule so the upsert picks up restatements; never treat a fresh day as final.
  • executemany is not atomic on its own. Idempotency comes from wrapping the upsert and the checkpoint in a single conn.transaction(). If you commit them separately, a crash between the two leaves a day loaded but unmarked, and the resume re-loads it — harmless with an upsert, but it defeats the skip optimization.
  • Let botocore’s built-in retry stay off. Setting max_attempts=0 keeps retry logic in one place. If both botocore and your loop retry, a throttled call is attempted far more often than your backoff schedule intends — the same double-retry trap covered in implementing retry logic for failed metric pulls.
  • GetCostAndUsageWithResources only reaches back 14 days. Resource-level granularity is capped at the last 14 days; for older ranges drop to get_cost_and_usage grouped by tag and linked account, and accept coarser attribution. Choose the API per chunk based on the day’s age.
  • Untagged spend must stay visible. Resources missing the tenant_id tag collapse into untagged rather than being dropped, so the backfill total always reconciles to the invoice. Tag-propagation lag means recently-created resources may land untagged for the day they were made.
  • Bump export_version only for deliberate restatements. Changing it mid-backfill forks every row into a new key and doubles your storage. Keep it fixed for a normal load; reserve a bump for when finance needs the pre- and post-correction figures preserved together.

Frequently Asked Questions

Why chunk by day instead of pulling the whole range in one call?

A day is the smallest unit Cost Explorer’s DAILY granularity commits cleanly, and it is the smallest unit you can safely re-run. Pulling a wide range in one call means a failure at 90% forces you to redo everything, and it makes the checkpoint coarse enough to be useless. Day chunks give a precise resume point and keep each transaction small enough to commit quickly under lock.

How is this different from just deleting and re-inserting the range?

Delete-then-insert has a window where the data is gone — any reader or downstream job that runs mid-backfill sees a hole, and a crash after the delete leaves the range empty. An upsert on the natural key never removes committed data; each day transitions atomically from its old values to its new ones, so the ledger is always complete and queryable during a resume.

What makes the re-run produce identical rows?

The natural-key upsert. Because (tenant_id, resource_id, billing_period, export_version) uniquely identifies a fact, a second write with the same key updates in place instead of inserting a duplicate, and Cost Explorer returns the same figure for a closed day every time. Combined with the atomic checkpoint, replaying any subset of days converges the table to exactly one correct state.

Where should record validation happen in this flow?

Between fetch_day and commit_day. Every parsed row should clear strict schema validation for billing data — non-numeric amounts, missing currency, or an out-of-range date — before it reaches the upsert, so a malformed API response is rejected rather than persisted as a bogus charge that a later re-run would faithfully reproduce.

Can I run several date ranges in parallel to finish faster?

Yes, if they use distinct job_id values and disjoint day ranges, because the upsert and checkpoint are keyed to avoid collisions. Watch the Cost Explorer request rate, though: parallel jobs share the account’s throttle budget, so more workers often just trade one long backfill for several throttled ones. Two or three parallel ranges is usually the practical ceiling.

Back to: Batch Processing for Historical Metrics