Incremental Models, Snapshots and Project Quality

37.1 Incremental Models (30 min)

A table model is rebuilt completely on every run. An incremental model processes only new or changed rows and merges them into the existing table. It's dlt's merge (Day 27), but inside the warehouse.

Is it worth it here?

Honestly, for most of our models, no: DuckDB rebuilds them in seconds. But the pattern matters as data grows (daily data for hundreds of series, or tick data), and it teaches you to think carefully about what "new" means. We'll build one good example: a fact table of every observation.

What counts as "new"? Use the load timestamp

Thanks to dlt's merge, a revised observation is rewritten with a new load ID (Day 27). So "rows loaded after my last run" captures both new observations and revisions. This is more robust than "dates after my latest date", which would miss revisions entirely.

Create models/marts/core/fct_observations.sql:

{{ config(
    materialized         = 'incremental',
    unique_key           = ['series_id', 'obs_date'],
    incremental_strategy = 'delete+insert',
    on_schema_change     = 'fail'
) }}

-- Grain: one row per series × observation date (all providers)
select
    {{ dbt_utils.generate_surrogate_key(['series_id', 'obs_date']) }}  as observation_key,
    series_id,
    series_name,
    provider,
    obs_date,
    value,
    loaded_at
from {{ ref('int_macro__observations') }}

{% if is_incremental() %}
where loaded_at > (select coalesce(max(loaded_at), timestamp '1900-01-01') from {{ this }})
{% endif %}

New Jinja pieces:

See it work

dbt run --select fct_observations                 # first run: builds the full table
dbt run --select fct_observations                 # second run: nothing new → inserts 0 rows
cd .. && python ingest.py && cd macro_dbt         # new data arrives
dbt run --select +fct_observations                # processes only newly loaded rows
cat target/run/macro_dbt/models/marts/core/fct_observations.sql

Read the executed SQL on an incremental run: a temporary batch, a delete of matching keys, then an insert.

Full refresh

When logic changes, rebuild from scratch:

dbt run --select fct_observations --full-refresh

Incremental pitfalls to remember: (1) rows deleted upstream are never deleted downstream; (2) late-arriving data with an old loaded_at would be missed; (3) any logic change needs --full-refresh. Add a test comparing row counts with the source (Try it 37.1). Our choice of loaded_at avoids pitfall 2 because dlt stamps every rewrite.

Try it 37.1

Write a singular test tests/assert_fct_observations_complete.sql that fails if fct_observations has a different number of rows than int_macro__observations.

with a as (select count(*) as n from {{ ref('fct_observations') }}),
     b as (select count(*) as n from {{ ref('int_macro__observations') }})
select a.n as fct_rows, b.n as source_rows
from a, b
where a.n <> b.n

37.2 Snapshots: Tracking Revisions Inside the Warehouse (20 min)

On Day 27 you kept vintages in dlt with an append-only table. dbt offers a warehouse-side alternative: snapshots, which record how a row changes over time (a type-2 slowly changing dimension).

Create snapshots/snap_cpi_national.sql:

{% snapshot snap_cpi_national %}

{{ config(
    target_schema = 'snapshots',
    unique_key    = 'snapshot_key',
    strategy      = 'check',
    check_cols    = ['cpi']
) }}

-- One row per national CPI component × month; a new version whenever the value changes
select
    month || '|' || component   as snapshot_key,
    month,
    component,
    cpi
from {{ ref('stg_statcan__cpi') }}
where geo_code = 'CA'

{% endsnapshot %}

dbt snapshot

dbt adds dbt_valid_from and dbt_valid_to columns. The first run records every row as valid from now. On later runs, a row whose cpi changed gets its old version closed (dbt_valid_to set) and a new version opened.

Find every revision ever observed:

select month, component, cpi, dbt_valid_from, dbt_valid_to
from snapshots.snap_cpi_national
where snapshot_key in (
    select snapshot_key from snapshots.snap_cpi_national
    group by snapshot_key having count(*) > 1
)
order by snapshot_key, dbt_valid_from;

Snapshots only see what exists when they run, so schedule dbt snapshot with every pipeline run (dbt build includes snapshots). Newer dbt versions also let you define snapshots in YAML. The block syntax shown here works across versions.

Why does this matter for economists? Forecast evaluation (Days 61–70) must use the data as it was known at the time, not today's revised figures. Snapshots let you reconstruct that.

37.3 Advanced DRY: Generating Staging Models, union_relations (23 min)

A macro for identical staging models

The three stg_<provider>__observations models differ only in the source name:

-- macros/stg_observations.sql
{% macro stg_observations(source_name) %}

with source as (

    select * from {{ source(source_name, 'observations') }}

)

select
    cast(date as date)       as obs_date,
    series_id,
    value,
    '{{ source_name }}'      as provider,
    _dlt_load_id             as load_id
from source

{% endmacro %}

Each model file becomes one line:

-- models/staging/boc/stg_boc__observations.sql
-- Grain: one row per BoC series × observation date
{{ stg_observations('boc') }}

Is this good DRY? Yes here: the three models share meaning (dlt's standard observation schema) and must change together. It would be bad DRY if one provider later needed special handling. Then you'd write that model out in full rather than add an if provider == ... branch to the macro.

dbt_utils.union_relations

Instead of hand-writing union all in int_macro__observations:

with unioned as (

    {{ dbt_utils.union_relations(relations=[
        ref('stg_boc__observations'),
        ref('stg_fred__observations'),
        ref('stg_statcan__observations')
    ]) }}

)

union_relations matches columns by name (like UNION ALL BY NAME, Day 12), fills missing columns with NULL, and adds a _dbt_source_relation column recording which model each row came from. Adding a fourth provider means adding one ref() to the list.

37.4 Operational Macros with run-operation (15 min)

Macros can also do things rather than generate model SQL: reports, maintenance, administration. Run them with dbt run-operation.

A staleness report

Create macros/ops/report_stale_series.sql:

{% macro report_stale_series() %}

    {% set query %}
        select
            s.series_name,
            cast(max(o.obs_date) as varchar)  as latest,
            s.max_staleness_days
        from {{ ref('seed_series') }} as s
        left join {{ ref('int_macro__observations') }} as o using (series_id)
        group by all
        having max(o.obs_date) is null
            or max(o.obs_date) < current_date - s.max_staleness_days
        order by 1
    {% endset %}

    {% set rows = run_query(query) %}
    {% if rows | length == 0 %}
        {{ log("✓ All series are within their staleness tolerance.", info=True) }}
    {% else %}
        {% for r in rows %}
            {{ log("⚠ STALE: " ~ r[0] ~ " (latest " ~ r[1] ~ ", tolerance "
                   ~ r[2] ~ " days)", info=True) }}
        {% endfor %}
    {% endif %}

{% endmacro %}

dbt run-operation report_stale_series

Cleaning up dev schemas

-- macros/ops/drop_dev_schemas.sql
{% macro drop_dev_schemas(dry_run=true) %}
    {% if target.name != 'dev' %}
        {{ exceptions.raise_compiler_error("drop_dev_schemas only runs on the dev target") }}
    {% endif %}
    {% set schemas = run_query(
        "select distinct schema_name from information_schema.schemata where schema_name like 'dev%'"
    ).columns[0].values() %}
    {% for s in schemas %}
        {% if dry_run %}
            {{ log("would drop schema " ~ s, info=True) }}
        {% else %}
            {% do run_query("drop schema if exists " ~ s ~ " cascade") %}
            {{ log("dropped schema " ~ s, info=True) }}
        {% endif %}
    {% endfor %}
{% endmacro %}

dbt run-operation drop_dev_schemas                        # dry run: lists only
dbt run-operation drop_dev_schemas --args '{dry_run: false}'
dbt build                                                  # rebuild dev from scratch

Notice the guard rails: the macro refuses to run outside dev, and it defaults to a dry run. Any operational macro that deletes things should have both.

{% do … %} runs an expression for its side effect without printing anything.

37.5 Project Quality (13 min)

Conventions, written down

Add macro_dbt/CONVENTIONS.md:

# macro_dbt conventions
- Layers: staging (views, 1 per source table) → intermediate (by concept) → marts (tables)
- Names: stg_<source>__<entity>, int_<concept>__<what>, fct_/dim_ for dimensional marts
- Only staging calls source(); marts never ref() staging directly
- Every model: grain comment at top + description in YAML
- Every model with a key: unique (or unique_combination_of_columns) + not_null tests
- Lowercase SQL, CTEs named as nouns, one column per line, trailing commas
- Constants in vars; series configuration in seeds; no hard-coded series lists
- Macros only for shared *meaning*; document every macro in _macros.yml

Optional: automatic SQL linting

pip install sqlfluff sqlfluff-templater-dbt

With a small .sqlfluff configuration (templater = dbt, dialect = duckdb), sqlfluff lint models/ checks style, and can check it automatically before each commit. It's optional, but common in professional dbt teams.

A review checklist for new models

• Correct layer and folder? Name follows conventions?
• Grain documented? Key tested for uniqueness?
• Reads only from the layer(s) it's allowed to?
• Materialisation justified?
• No duplicated logic that a macro or upstream model already provides?
• Regression-checked against the previous version, if it replaced something?

37.6 Capstone: A Complete, DRY, Tested Intermediate Layer (35 min)

Finish porting the remaining Module 2 intermediate logic, applying everything from Days 34–37.

Requirements:

1. int_cpi__smoothed: moving averages with the min_periods rule (Day 19). Write a macro rolling_avg(col, n_periods, partition_by, order_by, step='month') that returns the case when count(...) = n then avg(...) end expression over a range between interval (n-1) month preceding and current row window, and use it for yoy_ma3, yoy_ma12 and mom_ma3.
2. int_cpi__national_measures: port Day 19's model (headline, core ex food & energy, shelter/energy/food, median of majors, diffusion, contributions with residual, next-month-if-flat), using pct_change and ref('seed_cpi_weights').
3. int_gdp__cycle_flags: technical recession + overlaps_recession(quarter, quarter + 3 months − 1 day).
4. Tests: keys unique; is_first_of_period; within_range on headline YoY (−5, 20) and GDP growth; a unit test for rolling_avg proving that a 3-month average is NULL until 3 observations exist.
5. Incremental + snapshot: fct_observations and snap_cpi_national from 37.1–37.2 included in dbt build.
6. Regression: extend analyses/regression_vs_module2.sql with int_cpi__national_measures and int_gdp__cycle_flags, and get zero unexplained differences.
7. dbt build green; dbt run-operation report_stale_series runs; docs regenerated.

-- macros/rolling_avg.sql
{#
  rolling_avg(): trailing moving average over n periods, NULL until n observations exist
  (pandas' rolling(n, min_periods=n)). Uses a RANGE frame, so gaps in the date column
  shrink the window instead of silently widening it (Day 16).
#}
{% macro rolling_avg(col, n_periods, order_by, partition_by=[], step='month') -%}
    case
        when count({{ col }}) over (
            {% if partition_by %}partition by {{ partition_by | join(', ') }}{% endif %}
            order by {{ order_by }}
            range between interval {{ n_periods - 1 }} {{ step }} preceding and current row
        ) = {{ n_periods }}
        then avg({{ col }}) over (
            {% if partition_by %}partition by {{ partition_by | join(', ') }}{% endif %}
            order by {{ order_by }}
            range between interval {{ n_periods - 1 }} {{ step }} preceding and current row
        )
    end
{%- endmacro %}

-- models/intermediate/cpi/int_cpi__smoothed.sql
-- Grain: month × geography × component. Adds moving averages and 12-month acceleration.
select
    *,
    {{ rolling_avg('yoy', 3,  'month', ['geo', 'component']) }}   as yoy_ma3,
    {{ rolling_avg('yoy', 12, 'month', ['geo', 'component']) }}   as yoy_ma12,
    {{ rolling_avg('mom', 3,  'month', ['geo', 'component']) }}   as mom_ma3,
    yoy - lag(yoy, 12) over (partition by geo, component order by month) as yoy_change_12m
from {{ ref('int_cpi__growth') }}

unit_tests:
  - name: rolling_avg_respects_min_periods
    model: int_cpi__smoothed
    given:
      - input: ref('int_cpi__growth')
        rows:
          - {month: 2026-01-01, geo: Canada, component: All-items, yoy: 1.0, mom: 0.5}
          - {month: 2026-02-01, geo: Canada, component: All-items, yoy: 2.0, mom: 0.5}
          - {month: 2026-03-01, geo: Canada, component: All-items, yoy: 3.0, mom: 0.5}
    expect:
      rows:
        - {month: 2026-01-01, yoy_ma3: null}
        - {month: 2026-02-01, yoy_ma3: null}
        - {month: 2026-03-01, yoy_ma3: 2.0}

Recap: Days 34–37

Looking ahead: Days 38–40 — the macro data mart

• Join yield curves (2y, 5y, 10y GoC; term spreads) with USD/CAD, the BoC–Fed differential and oil to model an open-economy dashboard schema.
• Build macro_monthly, macro_quarterly and an open-economy mart with facts, dimensions and dashboard-ready panels.
• Cut over from Module 2 to dbt in prod, with a final regression test, and retire load_warehouse.py.

Day 37 checklist

• I can build an incremental model, choose a sound definition of "new", and know its pitfalls
• I can snapshot a model to record revisions over time
• I generate identical staging models from one macro and union them with union_relations
• I write operational macros with guard rails and run them with run-operation
• My project has written conventions and a review checklist
• The intermediate layer is complete, DRY, tested and regression-checked against Module 2

cd macro_dbt && dbt build && dbt docs generate && cd ..
git add . && git commit -m "Day 37: incremental, snapshots, advanced DRY, complete intermediate layer" && git push