Sources and freshness checks

Sources tell dbt where raw data lives. Freshness tells dbt when raw is too old to trust. Combined, they fail builds early when upstream is missing data, instead of silently producing dashboards built on stale extracts.

cloud_run_dbt/dbt/models/sources.yml
yaml
sources:
  - name: ecommerce_raw
    project: "{{ var('gcp_project_id') }}"
    dataset: "{{ var('bq_raw_dataset') }}"
    freshness:
      warn_after: { count: 1, period: day }
      error_after: { count: 2, period: day }
    loaded_at_field: loaded_at
    tables:
      - name: olist_orders_dataset
      - name: olist_order_items_dataset
      - name: olist_customers_dataset

Each source declares its table, freshness window, and the column that proves freshness. Builds fail if max(loaded_at) is older than the warn or error threshold.

cloud_run_dbt/app.py
bash
runner = dbtRunner()
cli_args = ["--project-dir", "dbt", "--profiles-dir", "dbt"]

logging.info("Running: dbt source freshness")
runner.invoke(["source", "freshness"] + cli_args)

logging.info("Running: dbt build")
result: dbtRunnerResult = runner.invoke(["build"] + cli_args)

The Cloud Run handler calls source freshness before build. If freshness fails, the build skips and the alert fires.

Add it. The simplest pattern: append , CURRENT_TIMESTAMP() AS loaded_at to the bq load select query. The Olist staging models in this course do exactly that. Freshness then checks the latest loaded_at value against the threshold.

Quiz: Quiz

Loading practice…