Configuring dbt source freshness for spatial feeds

This page walks through configuring dbt source freshness on a single spatial feed end to end: declaring the source with a trustworthy loaded_at_field, setting warn and error thresholds, running the check, scheduling it, and turning a breached threshold into an alert.

When to use this approach

Add source freshness to a geometry feed — rather than relying on downstream tests alone — when any of these hold:

• The feed can stop silently. GPS, AIS, and IoT sensor streams keep returning valid geometry long after they stop updating, so only a timestamp reveals the problem. If your source is a static reference layer that never changes, skip freshness and opt it out with freshness: null.
• Staleness is a data-quality incident, not a build error. Freshness catches a lagging input; it is the layer above the geometry-validity and SRID checks in spatial testing in CI pipelines. Use both — freshness gates the source, tests gate the models.
• You need an early, DAG-level gate. Freshness runs before models build, so it can abort a run before stale geometry propagates. Where it sits relative to everything else is covered in spatial model dependency graphs.

Prerequisites

• dbt Core 1.7+ with any warehouse adapter — freshness is engine-agnostic.
• An ingest timestamp column on the source table, stamped by your loader (not the device), stored in UTC. This example uses ingested_at.
• Grants to SELECT on the raw source schema so dbt can run max(ingested_at).
• A scheduler or CI runner able to invoke dbt at least as often as your tightest threshold.
• Connection secrets wired through dbt’s env_var() pattern, set once in profiles.yml:

# profiles.yml (excerpt)
dbt_geospatial:
  target: prod
  outputs:
    prod:
      type: postgres
      host: "{{ env_var('DBT_PG_HOST') }}"
      user: "{{ env_var('DBT_PG_USER') }}"
      password: "{{ env_var('DBT_PG_PASSWORD') }}"
      dbname: "{{ env_var('DBT_PG_DBNAME', 'analytics') }}"
      schema: staging

Step-by-step instructions

1. Declare the spatial source with a loaded_at_field

Create a source YAML that names the feed and points loaded_at_field at the ingest column. This example uses a real-time AIS position feed — high volume, prone to coverage gaps, and exactly the kind of source that stops silently.

# models/staging/_sources.yml
version: 2

sources:
  - name: ais
    database: raw
    schema: ingest
    tables:
      - name: vessel_positions
        description: "Decoded AIS position reports, one row per message."
        loaded_at_field: ingested_at   # warehouse ingest time, NOT the AIS timestamp
        columns:
          - name: geom
            description: "Vessel point, EPSG:4326."
          - name: ingested_at
            description: "UTC timestamp stamped by the loader on write."

Verify the source resolves and dbt can see the column:

dbt parse
dbt list --select source:ais.vessel_positions
# Expect the source to be listed with no parse errors.

2. Set warn and error thresholds matched to the feed cadence

Add a freshness block with warn_after and error_after. Each takes a count and a period of minute, hour, or day. Derive the numbers from the feed’s normal cadence — for a seconds-level AIS stream, warn at 15 minutes and error at 45, which tolerates coastal coverage gaps but treats a three-quarter-hour hole as a fault.

      - name: vessel_positions
        description: "Decoded AIS position reports, one row per message."
        loaded_at_field: ingested_at
        freshness:
          warn_after: {count: 15, period: minute}
          error_after: {count: 45, period: minute}
          # Scan only recent partitions so max() stays cheap on a huge feed.
          filter: ingested_at > current_timestamp - interval '1 day'

The filter clause keeps the freshness query from scanning the feed’s entire history — critical on high-volume geometry sources, and the same concern that drives handling large geospatial datasets.

The two thresholds carve the feed’s lag into three zones — the diagram below plots the age of the newest geometry over a day where the feed stalls late in the afternoon:

Verify the thresholds parse and the block is attached to the right table:

dbt parse   # a malformed period or missing loaded_at_field fails here

3. Run dbt source freshness

Run the check against the source. dbt issues the filtered max(ingested_at) query, computes the lag, and reports a status per table.

dbt source freshness --select source:ais.vessel_positions

A healthy run prints a PASS with the measured max_loaded_at and age; a lagging feed prints WARN or ERROR. The command also writes target/sources.json for programmatic use.

Verify the reported age is a small, plausible number on a live feed, and inspect the machine-readable output:

cat target/sources.json | python -m json.tool | grep -E '"status"|"max_loaded_at"'
# Expect "status": "pass" and a max_loaded_at within the last few minutes.

4. Wire the check into a scheduled job

Freshness is only useful if it runs on a cadence tighter than the threshold it enforces. Schedule it as its own step whose exit code fails the job. This GitHub Actions workflow runs every 15 minutes.

# .github/workflows/source-freshness.yml
name: source-freshness
on:
  schedule:
    - cron: "*/15 * * * *"   # match the tightest warn_after
  workflow_dispatch:
jobs:
  freshness:
    runs-on: ubuntu-latest
    env:
      DBT_PG_HOST: ${{ secrets.DBT_PG_HOST }}
      DBT_PG_USER: ${{ secrets.DBT_PG_USER }}
      DBT_PG_PASSWORD: ${{ secrets.DBT_PG_PASSWORD }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install dbt-postgres
      - run: dbt source freshness --select source:ais.vessel_positions

Because a breached error_after makes the command exit non-zero, the workflow run turns red on its own — no extra assertion needed.

Verify the schedule and exit-code behavior by triggering the workflow manually against a paused feed and confirming the run fails:

# Locally, simulate the CI step against a known-stale fixture:
dbt source freshness --select source:ais.stale_fixture; echo "exit=$?"
# Expect exit=1 (or another non-zero code) — proof the gate works.

5. Alert on lag by routing the freshness result

A red CI run is a start, but on-call wants a signal that names the feed and distinguishes a warning from an error. Parse target/sources.json and push a message, paging only on true errors.

# scripts/alert_freshness.py
import json, os, sys, urllib.request

with open("target/sources.json") as f:
    results = json.load(f)["results"]

errors = [r["unique_id"] for r in results if r["status"] == "error"]
warns  = [r["unique_id"] for r in results if r["status"] == "warn"]

if errors or warns:
    text = ""
    if errors:
        text += f":rotating_light: STALE feeds (error_after breached): {', '.join(errors)}\n"
    if warns:
        text += f":warning: lagging feeds (warn_after): {', '.join(warns)}"
    req = urllib.request.Request(
        os.environ["SLACK_WEBHOOK_URL"],
        data=json.dumps({"text": text}).encode(),
        headers={"Content-Type": "application/json"},
    )
    urllib.request.urlopen(req)

sys.exit(1 if errors else 0)   # page only on error, notify on warn

Add it as a final workflow step with if: always() so it runs even after the freshness command exits non-zero:

      - name: alert on freshness
        if: always()
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
        run: python scripts/alert_freshness.py

Verify end to end by pausing the feed in a sandbox, letting the schedule fire, and confirming a Slack message arrives naming source.ais.vessel_positions — then resume the feed and confirm the next run clears.

Configuration reference

Gotchas & edge cases

• Event time in loaded_at_field. Pointing at the AIS/GPS device timestamp instead of the ingest time lets a spoofed or backdated device clock report a dead feed as fresh. Always use the loader-stamped column.
• Naive local timestamps. Freshness compares against a UTC “now.” A local-time column drifts by your offset and fires false alarms near midnight — store ingested_at in UTC.
• No filter on a huge feed. Without filter, max(ingested_at) scans the entire history and the check can take longer than the window it protects. Restrict to the last day of partitions.
• Running freshness inside dbt build. It is not part of dbt build — it is a separate command. If your CI only runs dbt build, freshness never executes and the gate is silently absent.
• Re-emitted duplicates. A feed that reconnects and re-stamps old positions keeps max(loaded_at) fresh. Freshness detects “stopped writing,” not “writing stale duplicates” — pair it with a distinct-geometry test for the latter.

FAQ

Related

• Handling Large Geospatial Datasets — partitioning and filtering that keep the freshness max() query cheap.
• Spatial Model Dependency Graphs — where freshness sits above the DAG and gates the build.
• Spatial Testing in CI Pipelines — the model-level checks freshness runs alongside.

Up: Part of Source Freshness for Geometry Feeds.