01 - dbt: Staging Models

13 min read

dbt: Staging Models

Quote

“A pure task should be deterministic and idempotent, meaning that it will produce the same result every time it runs or re-runs.”

Source: Maxime Beauchemin | “Functional Data Engineering” (2018)

Staging Model Core Principles

RuleRationale
1:1 with source tableEasy to audit; changes in the source are immediately visible
Rename and cast onlyBusiness logic belongs in intermediate models
Materialise as viewsNo storage cost; always reflects current source data
Prefix stg_<source>__<entity>Makes origin instantly clear
One staging model per source tablePrevents hidden coupling between sources
Add _id surrogate key where natural key is complexSimplifies downstream joins

Related pattern

The rename-and-cast operations in staging models rely on the same sql-fundamentals patterns such as CAST, UPPER, TRIM, and COALESCE that appear throughout the SQL reference material.

No business logic in staging

If you find yourself writing a CASE WHEN that encodes a business rule such as “a return above 50% is suspicious,” that logic belongs in an intermediate model, not staging. Staging is for structural transformation only.

dbt _sources.yml — Full Declaration with Freshness

Current docs move freshness under config

Recent dbt docs place source freshness under a source or table config: block, and loaded_at_field has moved there as well. Legacy projects still show the older top-level pattern, but new notes and greenfield projects should prefer the current config-scoped syntax so examples match modern dbt behavior.

This source declaration centralizes raw-table freshness thresholds, boundary tests, and documentation so staging models can stay 1:1 with the source instead of embedding those checks in SQL.

# models/staging/market_data/_sources.yml
version: 2
 
sources:
  - name: market_data
    description: >
      End-of-day market data ingested from the primary exchange data vendor.
      Tables are loaded nightly after exchange close.
    database: raw_db
    schema: market_data_raw
    loaded_at_field: _ingested_at
    freshness:
      warn_after:  {count: 25, period: hour}
      error_after: {count: 49, period: hour}
 
    tables:
      - name: daily_prices
        description: "OHLCV data per security per trading day."
        freshness:
          warn_after:  {count: 6,  period: hour}
          error_after: {count: 24, period: hour}
        columns:
          - name: security_id
            description: "Vendor-assigned security identifier."
            tests: [not_null]
          - name: price_date
            description: "Calendar date of the price observation."
            tests: [not_null]
          - name: open_price
            tests: [not_null]
          - name: high_price
            tests: [not_null]
          - name: low_price
            tests: [not_null]
          - name: close_price
            tests:
              - not_null
              - dbt_utils.accepted_range:
                  min_value: 0
                  inclusive: false
          - name: adjusted_close_price
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  inclusive: false
          - name: volume
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  inclusive: true
 
      - name: corporate_actions
        description: "Dividends, splits, spin-offs, mergers."
        columns:
          - name: action_id
            tests: [not_null, unique]
          - name: security_id
            tests: [not_null]
          - name: action_type
            tests:
              - accepted_values:
                  values: ['DIVIDEND', 'SPLIT', 'SPINOFF', 'MERGER', 'RIGHTS']
          - name: effective_date
            tests: [not_null]
 
      - name: index_constituents
        description: "Point-in-time constituent membership with weights."
        columns:
          - name: index_id
            tests: [not_null]
          - name: security_id
            tests: [not_null]
          - name: effective_date
            tests: [not_null]
          - name: weight
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  max_value: 1
                  inclusive: true
 
  - name: esg
    description: "ESG scores and controversy data from third-party ESG provider."
    database: raw_db
    schema: esg_raw
    loaded_at_field: _ingested_at
    freshness:
      warn_after:  {count: 8,  period: day}
      error_after: {count: 15, period: day}
 
    tables:
      - name: scores
        description: "Environmental, Social, Governance pillar scores per security."
        columns:
          - name: security_id
            tests: [not_null]
          - name: score_date
            tests: [not_null]
          - name: esg_score
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  max_value: 100
          - name: e_score
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  max_value: 100
          - name: s_score
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  max_value: 100
          - name: g_score
            tests:
              - dbt_utils.accepted_range:
                  min_value: 0
                  max_value: 100

stg_market_data__daily_prices

This staging model standardizes prices, trading dates, exchange metadata, and ingestion metadata without changing business meaning or row grain.

-- models/staging/market_data/stg_market_data__daily_prices.sql
-- Materialisation inherited from project config: view, schema: silver
 
with source as (
 
    select * from {{ source('market_data', 'daily_prices') }}
 
),
 
renamed as (
 
    select
        -- Primary identifiers
        security_id                                      as security_id,
        cast(price_date as date)                         as price_date,
 
        -- OHLCV columns — standardised naming
        cast(open_price          as numeric)             as open_price,
        cast(high_price          as numeric)             as high_price,
        cast(low_price           as numeric)             as low_price,
        cast(close_price         as numeric)             as close_price,
        cast(adjusted_close_price as numeric)            as adjusted_close_price,
        cast(volume              as bigint)              as volume,
 
        -- Currency context
        upper(trim(currency_code))                       as currency_code,
 
        -- Exchange context
        upper(trim(exchange_code))                       as exchange_code,
 
        -- Metadata
        _ingested_at                                     as _ingested_at,
        _source_file                                     as _source_file
 
    from source
 
)
 
select * from renamed

stg_esg__scores

This ESG staging model normalizes score names, types, and provider metadata so downstream factor logic reads consistent columns regardless of raw provider quirks.

-- models/staging/esg/stg_esg__scores.sql
 
with source as (
 
    select * from {{ source('esg', 'scores') }}
 
),
 
renamed as (
 
    select
        -- Identifiers
        security_id                              as security_id,
        cast(score_date as date)                 as score_date,
 
        -- ESG pillar scores (0–100 scale, provider normalised)
        cast(esg_score as numeric(6,3))          as esg_score,
        cast(e_score   as numeric(6,3))          as e_pillar_score,
        cast(s_score   as numeric(6,3))          as s_pillar_score,
        cast(g_score   as numeric(6,3))          as g_pillar_score,
 
        -- Provider metadata
        upper(trim(provider_code))               as provider_code,
        upper(trim(rating_category))             as rating_category,
 
        -- Whether this is a restated score
        cast(is_restated as boolean)             as is_restated_flag,
 
        -- Ingestion metadata
        _ingested_at                             as _ingested_at
 
    from source
 
)
 
select * from renamed

stg_market_data__corporate_actions

This model preserves the raw corporate-action grain while adding a surrogate key that makes downstream joins and history tracking easier to manage.

-- models/staging/market_data/stg_market_data__corporate_actions.sql
 
with source as (
 
    select * from {{ source('market_data', 'corporate_actions') }}
 
),
 
renamed as (
 
    select
        -- Surrogate key (natural key is composite: security_id + action_type + effective_date)
        {{ dbt_utils.generate_surrogate_key([
            'security_id',
            'action_type',
            'effective_date'
        ]) }}                                    as corporate_action_key,
 
        -- Natural identifiers
        action_id                                as source_action_id,
        security_id                              as security_id,
        cast(effective_date as date)             as effective_date,
        cast(ex_date        as date)             as ex_date,
        cast(record_date    as date)             as record_date,
        cast(payment_date   as date)             as payment_date,
 
        -- Action classification
        upper(trim(action_type))                 as action_type,
 
        -- Split / dividend amounts
        cast(split_ratio         as numeric(12,6)) as split_ratio,
        cast(dividend_amount     as numeric(18,6)) as dividend_amount,
        upper(trim(dividend_currency))           as dividend_currency,
 
        -- Spinoff / merger target
        related_security_id                      as related_security_id,
 
        -- Ingestion metadata
        _ingested_at                             as _ingested_at
 
    from source
 
)
 
select * from renamed

stg_market_data__index_constituents

This constituent staging model keeps point-in-time membership rows source-aligned while standardizing dates, weights, and change-type metadata.

-- models/staging/market_data/stg_market_data__index_constituents.sql
 
with source as (
 
    select * from {{ source('market_data', 'index_constituents') }}
 
),
 
renamed as (
 
    select
        -- Surrogate key
        {{ dbt_utils.generate_surrogate_key([
            'index_id',
            'security_id',
            'effective_date'
        ]) }}                                    as constituent_key,
 
        -- Identifiers
        index_id                                 as index_id,
        security_id                              as security_id,
        cast(effective_date as date)             as effective_date,
 
        -- Weight (decimal; 0.05 = 5%)
        cast(weight as numeric(10,8))            as weight,
 
        -- Change type on this date
        upper(trim(change_type))                 as change_type,   -- ADD / REMOVE / REWEIGHT
 
        -- Ingestion metadata
        _ingested_at                             as _ingested_at
 
    from source
 
)
 
select * from renamed

_staging_market_data.yml — Column-Level Documentation

This YAML file documents published staging columns and declares the low-cost data-quality tests that should fail before bad raw data spreads deeper into the DAG.

# models/staging/market_data/_staging_market_data.yml
version: 2
 
models:
  - name: stg_market_data__daily_prices
    description: "Cleaned, renamed OHLCV data. 1:1 with raw daily_prices table."
    columns:
      - name: security_id
        description: "Vendor security identifier. Natural key."
        tests: [not_null]
      - name: price_date
        description: "Trading date of the observation."
        tests: [not_null]
      - name: close_price
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              inclusive: false
      - name: adjusted_close_price
        description: "Close price adjusted for corporate actions."
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [security_id, price_date, exchange_code]
 
  - name: stg_market_data__index_constituents
    description: "Point-in-time constituent membership per index."
    columns:
      - name: constituent_key
        tests: [not_null, unique]
      - name: weight
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [index_id, security_id, effective_date]
 
  - name: stg_market_data__corporate_actions
    columns:
      - name: corporate_action_key
        tests: [not_null, unique]
      - name: action_type
        tests:
          - accepted_values:
              values: ['DIVIDEND', 'SPLIT', 'SPINOFF', 'MERGER', 'RIGHTS']
 
  - name: stg_esg__scores
    columns:
      - name: esg_score
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100
    tests:
      - dbt_utils.unique_combination_of_columns:
          combination_of_columns: [security_id, score_date, provider_code]

dbt Source Freshness in Practice

These commands run source freshness globally or by source group so orchestration can stop downstream models when raw feeds are too stale for the SLA.

# Run freshness checks for all sources
dbt source freshness
 
# Check only the market_data source
dbt source freshness --select source:market_data
 
# Sample output:
# Found 2 sources
# Freshness of 2 sources checked in 0.43s
# ERROR Source market_data.daily_prices is 26 hours, 4 minutes out of date
# PASS  Source esg.scores is 2 days, 3 hours out of date (within threshold)

Wire freshness failures into your orchestration layer to block downstream runs when sources are stale.

Staging Model Anti-Patterns

Keep staging strictly source-aligned

Staging failures are usually boundary failures. Once a staging model joins another relation, applies an analytical filter, or hides structural drift with SELECT *, the raw-to-modeled contract stops being auditable and downstream debugging gets much harder.

Correct staging scope

Staging models should contain only column renames, type casts, UPPER or TRIM normalization, surrogate key generation, and ingestion metadata passthrough. Any join, filter, or business rule belongs in an intermediate model where it can be independently tested and documented.

Joining to other models: Staging models should reference only their own source. Any join introduces a dependency that belongs in the intermediate layer.

This wrong example leaks downstream dimensional context into a model that should stay 1:1 with its source table.

-- WRONG: join in staging
select p.*, s.company_name
from {{ source('market_data', 'daily_prices') }} p
left join {{ ref('dim_securities') }} s on p.security_id = s.security_id

Business logic and filters: Do not filter rows in staging unless the source truly contains structural garbage (e.g., empty header rows). Filtering valid data hides lineage.

This wrong example mixes analytical screening with structural cleanup, which makes the staging layer responsible for business semantics it cannot document or test safely.

-- WRONG: business filter in staging
where close_price > 0 and volume > 1000  -- this is analytical logic

Aggregation: Staging is never the right place to sum, average, or group. It is always 1:1 row-preserving.

Using SELECT * without aliasing: Even though staging is 1:1, always explicitly list columns. This makes schema drift immediately visible as a compilation error.

Graph View

Backlinks