Incremental Loading

27.1 Why Incremental? Five Loading Patterns (18 min)

Right now every run downloads everything since 2006, even though at most a few days have changed. That's slow, wastes the providers' bandwidth, and makes frequent refreshes (Days 51–60) impractical.

Two concepts make these work:

• a cursor: the column that tells you how far you've loaded (usually the observation date);
• state: somewhere to remember the cursor between runs. dlt stores it with the pipeline, and commits it only when a load succeeds.

27.2 dlt.sources.incremental: a Cursor on One Series (25 min)

Create day27/incremental_basics.py:

from datetime import date

import dlt
from dlt.sources.helpers import requests
from common import practice_pipeline

BASE = "https://www.bankofcanada.ca/valet"


@dlt.resource(name="overnight_rate", write_disposition="append", primary_key="date",
              columns={"value": {"data_type": "double"}})
def overnight_rate(obs_date=dlt.sources.incremental("date", initial_value=date(2024, 1, 1))):
    start = obs_date.start_value                  # where the previous run finished
    print(f"  requesting data from {start}")
    r = requests.get(f"{BASE}/observations/V39079/json",
                     params={"start_date": start.isoformat()}, timeout=30)
    r.raise_for_status()
    rows = []
    for obs in r.json().get("observations", []):
        try:
            rows.append({"date": date.fromisoformat(obs["d"]),
                         "value": float(obs["V39079"]["v"])})
        except (KeyError, TypeError, ValueError):
            continue
    yield rows


if __name__ == "__main__":
    p = practice_pipeline("incr_lab", "lab")
    info = p.run(overnight_rate())
    print(info)

Run it twice:

python day27/incremental_basics.py
python day27/incremental_basics.py
duckdb incr_lab.duckdb -c "SELECT _dlt_load_id, COUNT(*) FROM lab.overnight_rate GROUP BY 1"

• First run: requests from 2024-01-01 and loads everything since.
• Second run: requests from the last loaded date. dlt also filters out rows it already loaded at the boundary (using the primary key), so the second load adds almost nothing.

How it works

1. dlt.sources.incremental("date", initial_value=...) declares date as the cursor.
2. On each run, obs_date.start_value is the cursor's value from the last successful run (or initial_value the first time).
3. As rows pass through, dlt tracks the maximum date and saves it as the new state after a successful load.
4. The request uses start_value, so the API only sends new data. That's the real saving.

Where is the state?

import json
print(json.dumps(p.state.get("sources", {}), indent=2, default=str))

You'll find the cursor's last_value under the resource name. The state is also stored in the destination (_dlt_pipeline_state), so a pipeline can recover it even on a new machine.

27.3 Revisions: Lookback Windows + merge (20 min)

Economic data gets revised. StatCan revises recent months of many series with each release, and GDP is revised for years. A pure "only after the cursor" strategy never sees revisions to data it already loaded.

The standard remedy: re-request a window before the cursor, and merge on the primary key, so revised values overwrite old ones and new values are inserted.

         already loaded                 new
|----------------------------------|========|
                      |--lookback--|
                      ↑ request from here, merge on (series_id, date)

from datetime import timedelta


@dlt.resource(name="overnight_rate_rev", write_disposition="merge", primary_key="date",
              columns={"value": {"data_type": "double"}})
def overnight_rate_rev(obs_date=dlt.sources.incremental("date", initial_value=date(2024, 1, 1)),
                       lookback_days=30):
    start = obs_date.start_value - timedelta(days=lookback_days)
    ...  # same request/parse as before, using `start`

A subtlety. dlt's incremental filter drops rows older than the saved cursor, which would throw away the lookback rows you deliberately re-requested. Recent dlt versions offer a lag option on dlt.sources.incremental for exactly this. The approach in 27.4, managing the cursor yourself, avoids the issue entirely and works for many series at once, so it's what our capstone uses.

How long should the lookback be?

merge makes a generous lookback cheap and safe: rows that didn't change are simply rewritten with the same values.

27.4 Many Series, One Resource: Per-Series State (20 min)

Our observation resources load many series into one long table. A single cursor per resource is then wrong:

V39079   (daily)    last date: 2026-09-29
FEDFUNDS (monthly)  last date: 2026-08-01
cursor = max(...) = 2026-09-29  → next run skips FEDFUNDS September data!

The fix is to store one cursor per series in the resource's own state, a dictionary that dlt saves for you:

from datetime import date, timedelta

import dlt
from dlt.sources.helpers import requests

BASE = "https://www.bankofcanada.ca/valet"


@dlt.resource(name="boc_observations_incr", write_disposition="merge",
              primary_key=("series_id", "date"),
              columns={"value": {"data_type": "double"}})
def boc_observations_incr(series_ids, initial_date="2006-01-01", lookback_days=14):
    state = dlt.current.resource_state()                 # persisted dict for this resource
    cursors = state.setdefault("last_date", {})          # {series_id: "YYYY-MM-DD"}

    for sid in series_ids:
        if sid in cursors:
            start = date.fromisoformat(cursors[sid]) - timedelta(days=lookback_days)
        else:
            start = date.fromisoformat(initial_date)

        r = requests.get(f"{BASE}/observations/{sid}/json",
                         params={"start_date": start.isoformat()}, timeout=30)
        r.raise_for_status()

        rows = []
        for obs in r.json().get("observations", []):
            cell = obs.get(sid)
            try:
                rows.append({"date": date.fromisoformat(obs["d"]), "series_id": sid,
                             "value": float(cell.get("v")), "source": "boc"})
            except (AttributeError, TypeError, ValueError):
                continue

        if rows:
            cursors[sid] = max(row["date"] for row in rows).isoformat()
        print(f"  {sid}: from {start} → {len(rows)} rows")
        yield rows

Notes:

• dlt.current.resource_state() returns a dictionary that dlt saves only if the load succeeds. If the run crashes halfway, the cursors don't advance, so no data is skipped.
• Store JSON-friendly values (strings, numbers) in state. We save ISO date strings, not date objects.
• Because of merge, the lookback rows update existing rows instead of duplicating them.

Try it 27.4

Load ["V39079", "FXCADUSD"] into the practice pipeline, run it twice, and confirm: (a) the second run requests only recent data; (b) there are no duplicate (series_id, date) rows; (c) the state contains one cursor per series.

from common import practice_pipeline
p = practice_pipeline("incr_lab", "lab")
for _ in range(2):
    p.run(boc_observations_incr(["V39079", "FXCADUSD"]))

import json
print(json.dumps(p.state["sources"], indent=2, default=str))

SELECT series_id, date, COUNT(*) FROM lab.boc_observations_incr
GROUP BY ALL HAVING COUNT(*) > 1;          -- expect zero rows

27.5 Change Detection: Only Download Republished Tables (15 min)

Full StatCan tables (Days 10 and 18) are large, and most days they haven't changed. getCubeMetadata (Day 26) includes a release time. Store it in state and skip the download when it hasn't moved:

from datetime import datetime, timezone

import dlt
from dlt.sources.helpers import requests

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


@dlt.resource(name="statcan_releases", write_disposition="append")
def statcan_releases(product_ids):
    """Yield a row only for tables republished since the last successful run."""
    state = dlt.current.resource_state()
    seen = state.setdefault("release_time", {})

    r = requests.post(f"{WDS}/getCubeMetadata",
                      json=[{"productId": pid} for pid in product_ids], timeout=60)
    r.raise_for_status()

    for item in r.json():
        if item.get("status") != "SUCCESS":
            continue
        obj = item["object"]
        pid = str(obj.get("productId"))
        release = obj.get("releaseTime")
        if seen.get(pid) == release:
            print(f"  {pid}: unchanged since {release} — skipping")
            continue
        seen[pid] = release
        yield {"product_id": pid, "release_time": release,
               "detected_at": datetime.now(timezone.utc)}

Each row in statcan_releases records when a table was republished: a log of release events. On Days 58–60, Prefect will use exactly this signal to trigger the heavy table download and the dbt rebuild only when StatCan publishes something new.

27.6 Backfills, Resets and Vintages (13 min)

Backfilling a historical window

To (re)load a specific past period without disturbing the saved cursor, give the incremental an end_value:

p.run(overnight_rate(
    obs_date=dlt.sources.incremental("date", initial_value=date(2008, 1, 1),
                                     end_value=date(2010, 1, 1))))

When end_value is set, dlt treats the run as a backfill and does not update the state.

Starting over

p.run(source, refresh="drop_resources")    # drop the resources' tables AND state, then reload

Use this after changing a resource's logic in a way that invalidates what's already loaded, for example a different cleaning rule.

Keeping vintages: how did the data look then?

merge keeps only the latest value. For research on revisions (and for honest forecasting evaluation on Days 61–70), you may want every version. The simplest approach is a separate append table of snapshots:

@dlt.resource(name="gdp_vintages", write_disposition="append")
def gdp_vintages(latest_n=12):
    """Append the latest N quarters of real GDP on every run (one snapshot per run)."""
    ...  # POST getDataFromVectorsAndLatestNPeriods for your GDP vector; yield long rows

Each run's rows carry a different _dlt_load_id, so SQL can find revisions:

WITH v AS (
    SELECT g.date, g.value, l.inserted_at AS vintage
    FROM raw_dlt.gdp_vintages AS g
    JOIN raw_dlt._dlt_loads   AS l ON l.load_id = g._dlt_load_id
)
SELECT date,
       arg_min(value, vintage) AS first_estimate,
       arg_max(value, vintage) AS latest_estimate,
       arg_max(value, vintage) - arg_min(value, vintage) AS total_revision
FROM v
GROUP BY date
HAVING COUNT(DISTINCT value) > 1
ORDER BY date;

dlt also offers a built-in slowly changing dimension strategy, write_disposition={"disposition": "merge", "strategy": "scd2"}, which adds validity-period columns (_dlt_valid_from, _dlt_valid_to). It assumes each run delivers a complete snapshot, so read its documentation carefully before using it with windowed API requests.

27.7 Capstone: An Incremental macro_ingest (25 min)

Upgrade the Day 23 package.

Requirements:

1. BoC boc_observations: merge, per-series cursors (27.4), lookback_days configurable (default 14).
2. FRED fred_observations: merge, per-series cursors. The CSV endpoint always returns full history, so the saving here is in loading (only rows ≥ cursor − lookback are yielded), not downloading. Default lookback 180 days, for revisions.
3. StatCan statcan_observations: merge. Set latest_n in config to a revision-friendly window (e.g. 36 months) after an initial full load of 250.
4. Metadata resources stay replace.
5. Contracts from Day 25 stay in place.
6. New config keys in .dlt/config.toml:

[sources.boc]
series_ids = ["V39079", "FXCADUSD", "BD.CDN.2YR.DQ.YLD", "BD.CDN.10YR.DQ.YLD"]
start_date = "2006-01-01"
lookback_days = 14

[sources.fred]
series_ids = ["FEDFUNDS", "DCOILWTICO", "CANLOLITOAASTSAM"]
start_date = "2006-01-01"
lookback_days = 180

[sources.statcan]
vectors = [41690973, 2062815]
latest_n = 36

7. Because the tables change from replace to merge, do the first run with refresh="drop_resources" (add a --full-refresh flag to ingest.py).
8. Add an SQL test sql/tests/raw_dlt_no_duplicate_observations.sql (severity error).

"""Bank of Canada Valet source — incremental with per-series cursors."""
from datetime import date, timedelta

import dlt
from dlt.sources.helpers import requests

BASE = "https://www.bankofcanada.ca/valet"
OBS_CONTRACT = {"columns": "freeze", "data_type": "freeze"}


def _fetch(series_id, start):
    r = requests.get(f"{BASE}/observations/{series_id}/json",
                     params={"start_date": start}, timeout=30)
    r.raise_for_status()
    return r.json()


@dlt.source(name="boc", section="boc")
def boc_source(series_ids: list = dlt.config.value,
               start_date: str = dlt.config.value,
               lookback_days: int = 14):

    @dlt.resource(name="boc_observations", write_disposition="merge",
                  primary_key=("series_id", "date"),
                  columns={"value": {"data_type": "double"}},
                  schema_contract=OBS_CONTRACT)
    def observations():
        cursors = dlt.current.resource_state().setdefault("last_date", {})
        for sid in series_ids:
            start = (date.fromisoformat(cursors[sid]) - timedelta(days=lookback_days)
                     if sid in cursors else date.fromisoformat(start_date))
            rows = []
            for obs in _fetch(sid, start.isoformat()).get("observations", []):
                cell = obs.get(sid)
                try:
                    rows.append({"date": date.fromisoformat(obs["d"]), "series_id": sid,
                                 "value": float(cell.get("v")), "source": "boc"})
                except (AttributeError, TypeError, ValueError):
                    continue
            if rows:
                cursors[sid] = max(r["date"] for r in rows).isoformat()
            yield rows

    @dlt.resource(name="boc_series_metadata", write_disposition="replace",
                  primary_key="series_id",
                  schema_contract={"columns": "evolve", "data_type": "freeze"})
    def series_metadata():
        for sid in series_ids:
            info = _fetch(sid, date.today().isoformat()).get("seriesDetail", {}).get(sid, {})
            yield {"series_id": sid, "label": info.get("label"),
                   "description": info.get("description")}

    return observations, series_metadata

import sys

FULL_REFRESH = "--full-refresh" in sys.argv

# inside the loop over sources:
info = pipeline.run(make_source(),
                    refresh="drop_resources" if FULL_REFRESH else None)

-- severity: error
-- sql/tests/raw_dlt_no_duplicate_observations.sql
SELECT source, series_id, date, COUNT(*) AS n
FROM (SELECT source, series_id, date FROM raw_dlt.boc_observations
      UNION ALL SELECT source, series_id, date FROM raw_dlt.fred_observations
      UNION ALL SELECT source, series_id, date FROM raw_dlt.statcan_observations)
GROUP BY ALL
HAVING COUNT(*) > 1

Measure the improvement

time python ingest.py --full-refresh      # first, full load
time python ingest.py                     # incremental
python run_tests.py

-- How many rows did each run touch?
SELECT l.load_id, l.inserted_at, COUNT(o.series_id) AS boc_rows
FROM raw_dlt._dlt_loads AS l
LEFT JOIN raw_dlt.boc_observations AS o ON o._dlt_load_id = l.load_id
GROUP BY ALL
ORDER BY l.inserted_at DESC
LIMIT 5;

With merge, each row keeps the _dlt_load_id of the run that last wrote it. Rows inside the lookback window move to the newest load, and older rows keep their original load ID.

Recap: Days 24–27

Looking ahead: Days 28–30

You'll harden the BoC Valet and StatCan extractors for production: dlt's declarative REST API source and RESTClient, pagination and rate limits, failure scenarios and alerts. Then you'll switch the Day 18–20 staging models to read from raw_dlt, so the whole warehouse is fed by automated, incremental, contract-protected ingestion.

Day 27 checklist

• I can choose between full refresh, incremental append, incremental merge, change detection and snapshots
• I can use dlt.sources.incremental and inspect pipeline state
• I use lookback windows with merge to capture statistical revisions
• I manage per-series cursors with dlt.current.resource_state()
• I can backfill, reset and keep vintages
• macro_ingest now loads incrementally, with contracts, and passes a duplicates test

git add .
git commit -m "Day 27: incremental loading, revisions, change detection, vintages"
git push