Day 55: Caching · Applied Decision Intelligence & Data Architecture

Written lesson

Time	Block
0:00 – 0:23	55.1 What caching is, and how Prefect decides a cache hit
0:23 – 0:45	55.2 Caching an API fetch
0:45 – 0:53	Break
0:53 – 1:15	55.3 Cache keys from the data's own version
1:15 – 1:35	55.4 When not to cache
1:35 – 1:43	Break
1:43 – 2:10	55.5 Side-effect steps: skip by content fingerprint
2:10 – 2:30	55.6 Mini-project: a refresh that does nothing when nothing changed

55.1 What Caching Is (23 min)

Caching means: if this exact work was done before, reuse the stored result instead of doing it again. It's the same idea as dbt only rebuilding what changed (Day 31) or dlt's incremental cursors (Day 27), applied to Prefect tasks.

Prefect computes a cache key for each task run from a cache policy. If a previous run with the same key stored a result that hasn't expired, the task is skipped and enters a Completed state named Cached.

Cache policy	Key built from	Effect
`INPUTS`	the task's arguments	same inputs → cache hit
`TASK_SOURCE`	the task's source code	editing the function invalidates the cache
`RUN_ID`	the current flow run	only reuse within one run (e.g. flow retries)
`NONE`	nothing	never cache
`DEFAULT`	`INPUTS + TASK_SOURCE + RUN_ID`	Prefect's default: reuse only inside the same run

Policies combine with +. For cross-run caching, the usual choice is INPUTS + TASK_SOURCE: the same arguments and the same code.

Cached results are persisted to result storage, by default files under ~/.prefect/storage/, so they survive between runs.

55.2 Caching an API Fetch (23 min)

The CPI briefing (Day 51) fetches from StatCan on every run, though CPI only changes once a month. Cache it:

from datetime import timedelta

from prefect import task
from prefect.cache_policies import INPUTS, TASK_SOURCE


@task(cache_policy=INPUTS + TASK_SOURCE, cache_expiration=timedelta(hours=6))
def fetch_cpi(latest_n: int) -> list[tuple[date, float]]:
    ...                                              # unchanged from Day 51

Run the flow twice in a row:

python -m orchestration.flows.cpi_briefing
python -m orchestration.flows.cpi_briefing

The second run's fetch_cpi is Cached: no request to StatCan, and instant. Run it with latest_n=13: that's a different input, so it's a cache miss. Edit the task's code (even a comment): with TASK_SOURCE, that's also a miss.

Forcing a refresh

fetch_cpi.with_options(refresh_cache=True)(25)       # ignore any cached value; store a new one

or set the environment variable PREFECT_TASKS_REFRESH_CACHE=true for a whole run.

Try it 55.2

Why is cache_expiration=timedelta(hours=6) a better choice here than no expiration at all? What could go wrong with a 30-day expiration?

Answer

With INPUTS alone, the cache key doesn't know that StatCan published new data: latest_n=25 looks identical before and after a release. Without expiration, the briefing would keep showing stale CPI forever. A 6-hour expiration bounds the staleness to at most six hours after a release. A 30-day expiration could hide an entire monthly release. The next section fixes this properly by putting the data version into the key.

55.3 Cache Keys from the Data's Own Version (23 min)

Time-based expiration is a guess. Better: make the key depend on what the data provider says has changed. StatCan's getCubeMetadata reports each table's releaseTime (Day 27). If the release time hasn't changed, the data hasn't either.

from datetime import timedelta

import requests
from prefect import flow, task

WDS = "https://www150.statcan.gc.ca/t1/wds/rest"


@task(retries=2, retry_delay_seconds=30)
def cpi_release_time() -> str:
    r = requests.post(f"{WDS}/getCubeMetadata", json=[{"productId": 18100004}], timeout=30)
    r.raise_for_status()
    return r.json()[0]["object"]["releaseTime"]


def key_by_release(context, parameters) -> str:
    """Cache key: one entry per (CPI release, window length)."""
    return f"cpi-{parameters['release_time']}-{parameters['latest_n']}"


@task(cache_key_fn=key_by_release, cache_expiration=timedelta(days=45))
def fetch_cpi_for_release(release_time: str, latest_n: int) -> list:
    ...                                             # same fetch as before


@flow
def cpi_briefing_versioned(latest_n: int = 25):
    release = cpi_release_time()                    # cheap metadata call, every run
    points = fetch_cpi_for_release(release, latest_n)   # expensive fetch, only after a release
    ...

A cache_key_fn receives the run context and the task's parameters, and returns any string. Here the key contains the release time, so:

until StatCan republishes, every run is a cache hit;
the first run after a release is a miss and fetches the new data;
the 45-day expiration is just a safety net.

This is the orchestration counterpart of Day 30's change detection. The cheap question ("has it changed?") runs every time; the expensive work runs only when the answer is yes.

55.4 When Not to Cache (20 min)

Caching is powerful and dangerous. Only cache work whose result is fully determined by the cache key.

Don't cache…	Because…
steps with side effects (loading data, `dbt build`, publishing)	a cache hit returns the old result but doesn't redo the effect. If the effect was undone (a table dropped), nothing restores it
steps depending on hidden inputs (the current time, a database's current contents)	the key doesn't see them, so you get stale results
results containing secrets	they'd be written to result storage on disk
very large results (whole DataFrames of millions of rows)	storage fills up; loading them back can be slower than recomputing
non-deterministic work you want fresh (random sampling, "latest" queries)	it defeats the purpose

Where cached results live

ls ~/.prefect/storage | head
du -sh ~/.prefect/storage

Result files are serialised Python objects (by default using pickle). Treat the folder like data: don't commit it, and clear it if it grows too large (rm -rf ~/.prefect/storage/* resets all caches).

Try it 55.4

For each task, decide: cache or not? If yes, with which key? (a) fetch_valet(series_id) from the Day 52 concurrency demo; (b) publish_briefing() (Day 53 artifact); (c) a task computing a 10-year rolling correlation from a DataFrame argument; (d) run_policy_step("dbt_build", …).

Answers

(a) Yes, but with a short expiration (BoC data updates daily) or a key including today's date: INPUTS + TASK_SOURCE, expiring after a few hours. (b) No: it has a side effect (it creates an artifact) and reads the warehouse's current state, a hidden input. (c) Possibly: it's deterministic given its input, but hashing a large DataFrame for the key costs time, so only cache if the computation is expensive. (d) No, not with Prefect caching: it's a side-effect step. Use the fingerprint approach in 55.5 instead.

55.5 Side-Effect Steps: Skip by Content Fingerprint (28 min)

dbt build is our most expensive side-effect step. Re-running it when nothing in its inputs changed (a manual re-run, a retry of a later step, a weekend run) wastes time. We can't use result caching (55.4), but we can make an explicit decision:

Rebuild only if the warehouse's inputs or the dbt code changed since the last successful build.

What goes into a fingerprint?

Input	How we measure it
observation data	an order-independent hash of every `(series_id, date, value)` in the observation tables
full StatCan tables	the latest published release time (`raw_dlt.statcan_releases`)
dbt code and seeds	the Git commit plus any uncommitted changes in `macro_dbt/`

Why hash content rather than load IDs? Day 27's lookback windows rewrite recent rows on every run, which gives them new _dlt_load_ids even when their values are identical. A content hash changes only when the numbers change.

Create orchestration/tasks/fingerprints.py:

"""Content fingerprints: decide whether downstream work is needed."""
import hashlib
import subprocess

import duckdb
from prefect import get_run_logger, task

from orchestration.settings import ENV, REPO

OBSERVATION_TABLES = ["raw_dlt.boc_observations", "raw_dlt.fred_observations",
                      "raw_dlt.statcan_observations"]


@task(cache_policy=None)
def warehouse_fingerprint() -> str:
    """Fingerprint of dbt's inputs: observation content, StatCan releases, dbt code."""
    parts = []
    with duckdb.connect(ENV["MACRO_DUCKDB_PATH"], read_only=True) as con:
        for t in OBSERVATION_TABLES:
            h = con.sql(f"select bit_xor(hash(series_id, date, value)) from {t}").fetchone()[0]
            parts.append(f"{t}={h}")
        rel = con.sql("select max(release_time)::varchar from raw_dlt.statcan_releases").fetchone()[0]
        parts.append(f"releases={rel}")

    commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=REPO,
                            capture_output=True, text=True).stdout.strip()
    dirty = subprocess.run(["git", "diff", "HEAD", "--", "macro_dbt"], cwd=REPO,
                           capture_output=True, text=True).stdout
    parts += [f"commit={commit}", f"dirty={hashlib.sha256(dirty.encode()).hexdigest()}"]

    fp = hashlib.sha256("|".join(parts).encode()).hexdigest()[:16]
    get_run_logger().info(f"warehouse fingerprint {fp}")
    return fp

hash(series_id, date, value) is a DuckDB function returning a number per row. bit_xor combines them in an order-independent way, so the same rows in any order give the same fingerprint, and changing any value changes it.
git diff HEAD -- macro_dbt captures uncommitted edits, so a model edited but not yet committed still triggers a rebuild.
cache_policy=None makes sure the fingerprint itself is never cached. It must always look at the current data.

Remembering the last build: Prefect Variables

Variables are small named values stored by the Prefect server, readable and editable in the UI (Variables page):

from prefect.variables import Variable

Variable.set("dbt_last_fingerprint", fp, overwrite=True)
last = Variable.get("dbt_last_fingerprint", default=None)

The decision in the transform subflow

from prefect.states import Completed
from prefect.variables import Variable

from orchestration.tasks.fingerprints import warehouse_fingerprint


@flow(name="transform")
def transform_flow(drill: dict | None = None, force: bool = False) -> dict:
    logger = get_run_logger()
    fp = warehouse_fingerprint()
    last = Variable.get("dbt_last_fingerprint", default=None)

    if fp == last and not force:
        logger.info(f"Inputs unchanged since last build ({fp}) — skipping dbt")
        return {"dbt_build": Completed(name="Skipped", message=f"inputs unchanged ({fp})")}

    states = {"dbt_deps": step("dbt_deps", [DBT, "deps", "--quiet"], str(DBT_DIR), drill)}
    if outcome(states["dbt_deps"]) == "failed":
        return states
    states["dbt_build"] = step("dbt_build", [DBT, "build", "--target", "prod"], str(DBT_DIR), drill)
    if outcome(states["dbt_build"]) != "failed":
        Variable.set("dbt_last_fingerprint", fp, overwrite=True)      # only after success
        states["dbt_docs"] = step("dbt_docs", [DBT, "docs", "generate", "--target", "prod"],
                                  str(DBT_DIR), drill)
    return states

Add a flow parameter force_transform: bool = False to daily_refresh and pass it through.

Note the familiar rule, record the marker only after success (Day 30, Day 50): the fingerprint is saved only when the build succeeded, so a failed build is retried next run even if the inputs are the same.

Also update outcome() so that a state named Skipped counts as "ok". It's a Completed state, so the existing logic already treats it that way.

55.6 Mini-Project: A Refresh That Does Nothing When Nothing Changed (20 min)

Apply release-keyed caching to the CPI briefing (55.3), and a short-expiry cache to StatCan metadata calls used in the flow.
Implement the fingerprint skip (55.5) in daily_refresh.
Measure:

time python -c "from orchestration.flows.daily_refresh import daily_refresh; daily_refresh()"
time python -c "from orchestration.flows.daily_refresh import daily_refresh; daily_refresh()"

The second run (with no new data in between) should skip dbt. Then run with force_transform=True and confirm dbt runs.

Prove the fingerprint reacts to content: in a DuckDB session (with nothing else running), make a copy of raw_dlt.fred_observations, change one value in the original, run warehouse_fingerprint() from a Python shell, restore the original from the copy, and run it again. The fingerprint must change and then return to its previous value.

Weekend effect: the BoC and FRED don't publish daily financial data on weekends, so weekend runs typically find unchanged content, and the refresh completes in seconds, skipping dbt. You've just made the pipeline cheaper without scheduling exceptions.

Day 55 checklist

I understand cache keys, policies (INPUTS, TASK_SOURCE, RUN_ID, NONE), expiration and result storage
I cache pure fetches, and key them on the data's own version (release time) where possible
I never use result caching for side-effect steps
dbt build is skipped when a content fingerprint of its inputs and code is unchanged
Markers (fingerprints) are recorded only after success, and a force parameter overrides the skip

git add . && git commit -m "Day 55: caching, release-keyed caches, content fingerprints" && git push

Caching

What you'll learn today

Today's plan