Dbt Data Transformation

dbt Data Transformation — Agent Skill

Production-grade analytics engineering with dbt™ (data build tool). This skill covers the complete dbt workflow: model authoring, testing, documentation, semantic layer design, multi-project dbt Mesh governance, platform operations, and migration strategies.

When to Use This Skill

✅ DO trigger when:

• Create dbt model — "Create a dbt model that transforms raw orders into a customer_orders mart"
• Write dbt test — "Write data tests for my staging models to ensure referential integrity"
• dbt unit test — "Add unit tests for this incremental model's merge logic"
• Build semantic layer — "Define MetricFlow semantic models and metrics for revenue reporting"
• Migrate dbt — "Migrate our v1.5 project to v1.9 with best practices"
• dbt Mesh — "Set up cross-project refs between our finance and marketing dbt projects"
• MetricFlow — "Create a saved query for monthly recurring revenue"
• dbt project — "Initialize a new dbt project following best-practice structure"
• Debug dbt job — "Troubleshoot this failed incremental model build"
• Optimize dbt — "Reduce warehouse costs for our daily dbt run"
• Model contracts — "Enforce model contracts on our marts models"
• dbt documentation — "Generate or improve dbt docs for our models"
• Snapshots / SCD — "Configure snapshot strategies for slowly changing dimensions"
• Jinja macros — "Write a Jinja macro for dynamic schema generation"

❌ DO NOT trigger for:

• General SQL questions — "How do I write a SELECT statement?" → Use query-expert skill
• Data pipeline orchestration — "How do I schedule an Airflow DAG?" → Use pipeline/orchestration skills
• Snowflake/BigQuery admin — "How do I create a Snowflake warehouse?" → Use warehouse-platform skills
• General ETL/ELT design — "Design our data ingestion pipeline" → Only if the answer is dbt-specific
• Python data processing — "Write a pandas script to clean data" → Use python or data-analysis skill

Progressive Disclosure — Domain Selection

This skill is organized into 5 domains. At the start of each interaction, assess which domain the user needs and disclose only the relevant depth.

Domain Map

Level 1: Analytics Engineering (always visible)
  └─ Model patterns, project structure, materializations, Jinja, sources, seeds
Level 2: Testing (disclose on test/singular/unit/generic trigger)
  └─ Data tests, unit tests, singular tests, custom generic tests, severity thresholds, Great Expectations
Level 3: Semantic Layer (disclose on MetricFlow/semantic/metric trigger)
  └─ MetricFlow, semantic models, metrics, dimensions, measures, saved queries
Level 4: dbt Mesh (disclose on cross-project/governance/contract trigger)
  └─ Cross-project refs, model contracts, access controls, groups, versions, governance
Level 5: Platform & Operations (disclose on job/CLI/migration/debug trigger)
  └─ Job troubleshooting, dbt MCP server, CLI commands, Fusion migration, warehouse optimization

Trigger → Domain Routing

Level 1: Analytics Engineering

Project Structure (Best Practice)

my_dbt_project/
├── dbt_project.yml
├── packages.yml
├── macros/
│   ├── cross_db_utils.sql
│   └── generate_schema_name.sql
├── models/
│   ├── staging/
│   │   └── source_name/
│   │       ├── _source_name__models.yml
│   │       ├── stg_source_name__table1.sql
│   │       └── stg_source_name__table2.sql
│   ├── intermediate/
│   │   └── int_customer_metrics.sql
│   └── marts/
│       ├── finance/
│       │   ├── _finance__models.yml
│       │   └── fct_orders.sql
│       └── marketing/
│           └── dim_customers.sql
├── seeds/
│   └── country_codes.csv
├── snapshots/
│   └── products_snapshot.sql
├── tests/
│   ├── generic/
│   │   └── assert_positive_value.sql
│   └── singular/
│       └── check_order_totals.sql
└── analyses/
    └── customer_cohorts.sql

Naming Conventions

Materialization Decision Matrix

                    ┌──────────────────────────────────────────────┐
                    │           CHOOSE MATERIALIZATION              │
                    └──────────────────────────────────────────────┘
                                        │
                        ┌───────────────┼───────────────┐
                        ▼               ▼               ▼
                   View/Ephemeral    Table         Incremental
                        │               │               │
         ┌──────────────┤       ┌───────┤       ┌───────┤
         ▼              ▼       ▼               ▼
    Staging        Intermediate  Marts (fct/dim)  Large fact tables
    Light transforms  Joins     Small-medium    Append-only or
    No persistence    Chained   Snapshot        merge-update
    needed            deps     dimensions       patterns

Rules of thumb:

• View: Staging models, light transformations, always-fresh data needed. Zero storage cost.
• Ephemeral: Intermediate models only referenced by 1-2 downstream models. Avoids view stacking without table write cost.
• Table: Marts models (fct/dim), snapshot dimensions, models referenced by many downstream consumers. Rebuilt fully on each run.
• Incremental: Large fact tables (>1M rows), append-only pipelines, event data. Must define is_incremental() block and unique_key.

Model Patterns

Staging Model

-- models/staging/stripe/stg_stripe__payments.sql
WITH source AS (
    SELECT * FROM {{ source('stripe', 'payments') }}
),

renamed AS (
    SELECT
        id                  AS payment_id,
        order_id,
        payment_method,
        status,
        amount              / 100.0 AS amount_dollars,   -- cents → dollars
        currency,
        created_at          AS payment_created_at,
        _batched_at         AS _batched_at
    FROM source
)

SELECT * FROM renamed

# models/staging/stripe/_stripe__models.yml
version: 2

models:
  - name: stg_stripe__payments
    description: >
      Staged Stripe payment data. One row per payment transaction.
      Amounts are converted from cents to dollars.
    columns:
      - name: payment_id
        description: Primary key from Stripe.
        data_tests:
          - unique
          - not_null
      - name: order_id
        description: Foreign key to orders.
        data_tests:
          - not_null
          - relationships:
              to: ref('stg_jaffle_shop__orders')
              field: order_id
      - name: payment_method
        data_tests:
          - accepted_values:
              values: ['credit_card', 'debit_card', 'bank_transfer', 'gift_card']
      - name: amount_dollars
        description: Payment amount in USD (converted from cents).
        data_tests:
          - not_null
      - name: payment_created_at
        data_tests:
          - not_null

Intermediate Model

-- models/intermediate/int_customer_payments.sql
{{
    config(
        materialized = 'ephemeral'
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_jaffle_shop__orders') }}
),

payments AS (
    SELECT * FROM {{ ref('stg_stripe__payments') }}
),

customer_payments AS (
    SELECT
        orders.customer_id,
        orders.order_id,
        SUM(payments.amount_dollars) AS total_amount,
        COUNT(DISTINCT payments.payment_id) AS payment_count,
        MAX(payments.payment_created_at) AS last_payment_at
    FROM orders
    LEFT JOIN payments USING (order_id)
    GROUP BY 1, 2
)

SELECT * FROM customer_payments

Fact Mart (Incremental)

-- models/marts/finance/fct_orders.sql
{{
    config(
        materialized = 'incremental',
        unique_key = 'order_id',
        incremental_strategy = 'merge',
       
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_jaffle_shop__orders') }}
    {% if is_incremental() %}
        WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
    {% endif %}
),

payments AS (
    SELECT * FROM {{ ref('int_customer_payments') }}
),

final AS (
    SELECT
        orders.order_id,
        orders.customer_id,
        orders.order_date,
        orders.status,
        orders.updated_at,
        COALESCE(payments.total_amount, 0) AS total_amount,
        COALESCE(payments.payment_count, 0) AS payment_count,
        payments.last_payment_at
    FROM orders
    LEFT JOIN payments USING (order_id)
)

SELECT * FROM final

Dimension Model (SCD Type 2 via Snapshot)

-- snapshots/products_snapshot.sql
{% snapshot products_snapshot %}

{{
    config(
        target_database = 'analytics',
        target_schema = 'snapshots',
        unique_key = 'product_id',
        strategy = 'timestamp',
        updated_at = 'updated_at',
        invalidate_hard_deletes = True
    )
}}

SELECT * FROM {{ source('raw', 'products') }}

{% endsnapshot %}

Source Configuration

# models/sources.yml
version: 2

sources:
  - name: stripe
    database: raw
    schema: stripe_data
    description: Stripe payment processing data loaded by Fivetran.
    loader: fivetran
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _batched_at

    tables:
      - name: payments
        description: Raw payment transactions.
        columns:
          - name: id
            data_tests:
              - unique
              - not_null

      - name: refunds
        description: Raw refund transactions.

      - name: customers

Jinja Macro Patterns

Dynamic Schema Generation

-- macros/generate_schema_name.sql
{% macro generate_schema_name(custom_schema_name, node) -%}

    {%- set default_schema = target.schema -%}

    {%- if custom_schema_name is none -%}
        {{ default_schema }}
    {%- elif target.name == 'prod' -%}
        {{ custom_schema_name | trim }}
    {%- else -%}
        {{ default_schema }}_{{ custom_schema_name | trim }}
    {%- endif -%}

{%- endmacro %}

Custom Generic Test

-- macros/positive_values.sql
{% test positive_values(model, column_name) %}

    SELECT *
    FROM {{ model }}
    WHERE {{ column_name }} < 0

{% endtest %}

Multi-Database Date Spine

-- macros/date_spine.sql
{% macro date_spine(datepart, start_date, end_date) %}
    {% if target.type == 'bigquery' %}
        SELECT * FROM UNNEST(
            GENERATE_DATE_ARRAY(
                DATE('{{ start_date }}'),
                DATE('{{ end_date }}'),
                INTERVAL 1 {{ datepart }}
            )
        ) AS date_day
    {% elif target.type == 'snowflake' %}
        -- Snowflake implementation
        SELECT DATEADD({{ datepart }}, ROW_NUMBER() OVER (ORDER BY SEQ4()) - 1,
            '{{ start_date }}'::DATE) AS date_day
        FROM TABLE(GENERATOR(ROWCOUNT => {{ dbt_utils.pretty_time() }}))
    {% endif %}
{% endmacro %}

Seeds

-- seeds/country_codes.csv
country_code,country_name,region,sub_region
US,United States,Americas,North America
GB,United Kingdom,Europe,Northern Europe
DE,Germany,Europe,Western Europe

# dbt_project.yml
seeds:
  my_project:
    country_codes:
      +column_types:
        country_code: varchar(2)
        country_name: varchar(100)
        region: varchar(50)
        sub_region: varchar(50)

dbt_project.yml Best Practice

name: 'my_analytics_project'
version: '1.0.0'
config-version: 2
require-dbt-version: [">=1.9.0", "<2.0.0"]

profile: 'my_project'

model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
asset-paths: ["assets"]

clean-targets:
  - "target"
  - "dbt_packages"
  - "logs"

flags:
  send_anonymous_usage_stats: False

vars:
  # Project-level variables
  surrogate_key_treat_nulls_as_empty_strings: True
  'dbt_date:time_zone': 'UTC'

models:
  my_analytics_project:
    staging:
      +materialized: view
      +schema: staging
      +tags: ['staging']
    intermediate:
      +materialized: ephemeral
      +schema: intermediate
      +tags: ['intermediate']
    marts:
      +materialized: table
      +schema: marts
      +tags: ['marts']
      finance:
        +schema: finance
      marketing:
        +schema: marketing

snapshots:
  my_analytics_project:
    +target_schema: snapshots

seeds:
  my_analytics_project:
    +schema: seeds

Level 2: Testing

Test Strategy Pyramid

                    ┌──────────────┐
                    │   Unit Tests  │  ← Isolated logic (Jinja, SQL logic)
                    │   (specific)  │
                    ├───────────────┤
                    │  Data Tests   │  ← Column-level (unique, not_null, etc.)
                    │  (generic)    │
                    ├───────────────┤
                    │ Singular Tests│  ← Complex business rules
                    │  (one-off)    │
                    ├───────────────┤
                    │  Source       │  ← Freshness + row count
                    │  Freshness    │
                    └───────────────┘

Data Tests (Schema YAML)

The most common pattern — declared inline in model YAML files:

models:
  - name: fct_orders
    columns:
      - name: order_id
        data_tests:
          - unique
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"
      - name: customer_id
        data_tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: total_amount
        data_tests:
          - not_null
          - positive_values   # custom generic test
      - name: status
        data_tests:
          - not_null
          - accepted_values:
              values: ['pending', 'shipped', 'delivered', 'cancelled']
              config:
                severity: error

Custom Generic Tests

Generic tests are reusable test functions defined as SQL files in tests/generic/.

-- tests/generic/assert_referential_integrity.sql
{% test assert_referential_integrity(model, column_name, to, field) %}

    SELECT {{ column_name }}
    FROM {{ model }}
    WHERE {{ column_name }} IS NOT NULL
    EXCEPT
    SELECT {{ field }}
    FROM {{ to }}

{% endtest %}

Usage in YAML:

columns:
  - name: customer_id
    data_tests:
      - assert_referential_integrity:
          to: ref('dim_customers')
          field: customer_id

Singular Tests

One-off SQL queries that return rows when a test fails:

-- tests/singular/check_order_total_consistency.sql
-- Orders with items should have total_amount > 0
-- Orders with total_amount = 0 should have no items

WITH order_items AS (
    SELECT order_id, COUNT(*) AS item_count
    FROM {{ ref('stg_jaffle_shop__order_items') }}
    GROUP BY 1
),

orders AS (
    SELECT order_id, total_amount
    FROM {{ ref('fct_orders') }}
)

SELECT
    o.order_id,
    o.total_amount,
    COALESCE(oi.item_count, 0) AS item_count
FROM orders o
LEFT JOIN order_items oi USING (order_id)
WHERE
    -- Fail: has items but amount is 0
    (COALESCE(oi.item_count, 0) > 0 AND o.total_amount <= 0)
    OR
    -- Fail: no items but amount > 0
    (COALESCE(oi.item_count, 0) = 0 AND o.total_amount > 0)

Unit Tests (dbt v1.8+)

Unit tests validate model transformation logic in isolation — no warehouse data needed.

# models/marts/finance/_finance__models.yml
unit_tests:
  - name: test_fct_orders_incremental_merge
    description: >
      Verify incremental merge logic: existing rows are updated,
      new rows are inserted, unchanged rows are preserved.
    model: fct_orders

    given:
      # Existing data in the incremental model
      - input: ref('stg_jaffle_shop__orders')
        rows:
          - {order_id: 1, customer_id: 100, status: 'pending', updated_at: '2024-01-01'}
          - {order_id: 3, customer_id: 102, status: 'shipped', updated_at: '2024-01-01'}
      - input: ref('int_customer_payments')
        rows:
          - {order_id: 1, customer_id: 100, total_amount: 50, payment_count: 1}
          - {order_id: 3, customer_id: 102, total_amount: 75, payment_count: 1}

      # Format for existing model data in incremental context
      - input: this
        rows:
          - {order_id: 1, customer_id: 100, total_amount: 50, status: 'pending', updated_at: '2024-01-01'}
          - {order_id: 2, customer_id: 101, total_amount: 100, status: 'shipped', updated_at: '2024-01-01'}

    expect:
      rows:
        # order_id=1: updated in source, should be overwritten
        - {order_id: 1, customer_id: 100, total_amount: 50, status: 'pending'}
        # order_id=2: not in source, should be preserved
        - {order_id: 2, customer_id: 101, total_amount: 100, status: 'shipped'}
        # order_id=3: new in source, should be inserted
        - {order_id: 3, customer_id: 102, total_amount: 75, status: 'shipped'}

Test Severity Configuration

# dbt_project.yml — global test severity defaults
tests:
  my_project:
    +severity: warn     # default for all tests

models:
  my_project:
    marts:
      +severity: error  # marts tests are errors

# Override per-test in YAML
  - name: fct_orders
    columns:
      - name: order_id
        data_tests:
          - unique:
              severity: error     # this one must pass
          - not_null:
              severity: error

Great Expectations Integration

For advanced data quality needs beyond dbt native tests:

# Install the dbt-expectations package
# Add to packages.yml:
packages:
  - package: calogica/dbt_expectations
    version: ">=0.10.0,<0.11.0"

# Usage in model YAML
models:
  - name: fct_orders
    columns:
      - name: total_amount
        data_tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000
              row_condition: "status != 'cancelled'"
              strictly: false
      - name: status
        data_tests:
          - dbt_expectations.expect_column_values_to_be_in_set:
              value_set: ['pending', 'shipped', 'delivered', 'cancelled']
      - name: payment_count
        data_tests:
          - dbt_expectations.expect_column_values_to_be_of_type:
              column_type: integer

Source Freshness

sources:
  - name: stripe
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
      filter: "DATE(created_at) >= CURRENT_DATE - 7"
    loaded_at_field: _batched_at

Run freshness checks:

dbt source freshness          # All sources
dbt source freshness --select source:stripe.payments  # Specific source

Level 3: Semantic Layer

Overview

The dbt Semantic Layer, powered by MetricFlow, transforms dbt models into a governed metrics platform. Semantic models define entities, dimensions, and measures in YAML — then MetricFlow generates optimized SQL for any metric query across any combination of dimensions.

Semantic Model

# models/marts/finance/_semantic_models.yml
semantic_models:
  - name: orders
    description: Order-level revenue and transaction data.
    model: ref('fct_orders')

    defaults:
      agg_time_dimension: order_date

    entities:
      - name: order
        type: primary
        expr: order_id
      - name: customer
        type: foreign
        expr: customer_id
      - name: location
        type: foreign
        expr: location_id

    dimensions:
      - name: order_date
        type: time
        type_params:
          time_granularity: day
        expr: order_date
      - name: order_status
        type: categorical
        expr: status
      - name: is_first_order
        type: categorical
        expr: CASE WHEN order_number = 1 THEN TRUE ELSE FALSE END
      - name: order_year
        type: time
        type_params:
          time_granularity: year
        expr: DATE_TRUNC('year', order_date)

    measures:
      - name: revenue
        description: Total gross revenue before discounts.
        agg: sum
        expr: total_amount
        create_metric: true
      - name: order_count
        description: Count of distinct orders.
        agg: count_distinct
        expr: order_id
        create_metric: true
      - name: average_order_value
        description: Average revenue per order.
        agg: average
        expr: total_amount
      - name: total_discounts
        agg: sum
        expr: discount_amount
      - name: unique_customers
        agg: count_distinct
        expr: customer_id

Metrics

# models/marts/finance/_metrics.yml
metrics:
  - name: revenue
    description: Total gross revenue across all orders.
    label: Revenue
    type: simple
    type_params:
      measure: revenue

  - name: average_order_value
    description: Average basket value per order.
    label: Avg. Order Value
    type: simple
    type_params:
      measure: average_order_value

  - name: rolling_7d_revenue
    description: Rolling 7-day revenue for trend analysis.
    label: 7-Day Revenue
    type: simple
    type_params:
      measure: revenue
    filter: |
      {{ TimeDimension('order_date', 'day') }} >= DATEADD('day', -7, CURRENT_DATE())

  - name: revenue_growth_pct
    description: Period-over-period revenue growth rate.
    label: Revenue Growth %
    type: derived
    type_params:
      expr: (revenue - revenue_prev_period) / NULLIF(revenue_prev_period, 0) * 100
      metrics:
        - name: revenue
        - name: revenue
          offset_window: 1 period
          alias: revenue_prev_period

  - name: customer_lifetime_value
    description: Average total revenue per unique customer.
    label: Customer LTV
    type: ratio
    type_params:
      numerator: revenue
      denominator: unique_customers

  - name: discount_rate
    description: Discounts as a percentage of gross revenue.
    label: Discount Rate %
    type: ratio
    type_params:
      numerator: total_discounts
      denominator: revenue

Saved Queries

# models/marts/finance/_saved_queries.yml
saved_queries:
  - name: monthly_revenue_by_status
    description: Monthly revenue broken down by order status.
    label: Monthly Revenue by Status
    query_params:
      metrics:
        - revenue
      group_by:
        - TimeDimension('order_date', 'month')
        - Dimension('order_status')
      order_by:
        - TimeDimension('order_date', 'month')
      where:
        - "{{ Dimension('order_status') }} != 'cancelled'"

  - name: top_customers_ltv
    description: Top 25 customers by lifetime value.
    label: Top 25 Customer LTV
    query_params:
      metrics:
        - customer_lifetime_value
      group_by:
        - Dimension('customer_name')
      order_by:
        - Metric('customer_lifetime_value', descending=true)
      limit: 25

  - name: weekly_revenue_trend
    description: Weekly revenue trend with 7-day rolling and growth rate.
    label: Weekly Revenue Trend
    query_params:
      metrics:
        - revenue
        - rolling_7d_revenue
        - revenue_growth_pct
      group_by:
        - TimeDimension('order_date', 'week')
      order_by:
        - TimeDimension('order_date', 'week')

Semantic Layer Commands

# Validate semantic manifests
dbt parse
mf validate-configs          # MetricFlow config validation

# List available entities
mf list entities

# List metrics
mf list metrics

# Query metrics via CLI
mf query --metrics revenue --group-by metric_time__week

# Query with filter
mf query --metrics revenue,order_count \
  --group-by metric_time__month \
  --where "{{ Dimension('order_status') }} = 'shipped'"

# Export query to various formats
mf query --metrics revenue --group-by metric_time__day --explain

MetricFlow Time Dimensions

When querying metrics, MetricFlow automatically creates metric_time as the canonical time dimension:

# Primary time grain from semantic model defaults
mf query --metrics revenue --group-by metric_time

# Specific grain
mf query --metrics revenue --group-by metric_time__day
mf query --metrics revenue --group-by metric_time__week
mf query --metrics revenue --group-by metric_time__month
mf query --metrics revenue --group-by metric_time__quarter
mf query --metrics revenue --group-by metric_time__year

Level 4: dbt Mesh — Multi-Project Governance

Architecture

dbt Mesh enables enterprise-scale data transformation by splitting a monolithic project into coordinated sub-projects that share models through governed interfaces.

┌─────────────────────────────────────────────────────────┐
│                    dbt Mesh Architecture                 │
├─────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐│
│ Source  │  │ Staging  │  │  Marts   │  │  Downstream  ││
│ Project │→│ Project  │→│  Project │→│   Consumer    ││
│         │  │          │  │          │  │   Projects    ││
├─────────┤  ├──────────┤  ├──────────┤  ├──────────────┤│
│ Raw data│  │ Standard │  │ Business │  │ BI / ML /    ││
│ sources │  │ ized     │  │ logic    │  │ Reverse ETL  ││
│         │  │ models   │  │ models   │  │              ││
└─────────┘  └──────────┘  └──────────┘  └──────────────┘│
     │              │              │              │        │
     └──────────────┴──────────────┴──────────────┘        │
                    Cross-Project Refs                     │
                    Model Contracts                        │
                    Access Controls                        │
                    Versioning                             │
└─────────────────────────────────────────────────────────┘

Cross-Project Refs

In a downstream project's dependencies.yml:

# dependencies.yml — in the downstream/consumer project
projects:
  - name: finance_marts
    description: Finance department dbt models.

dependencies:
  - project: finance_marts

Then reference models from the upstream project using three-part refs:

-- models/marts/marketing/fct_marketing_attribution.sql
-- Cross-project ref to the finance project's orders model
SELECT
    campaign_id,
    SUM({{ ref('finance_marts', 'fct_orders') }}.total_amount) AS attributed_revenue
FROM {{ ref('stg_marketing__campaigns') }}
LEFT JOIN {{ ref('finance_marts', 'fct_orders') }}
    ON campaigns.order_id = {{ ref('finance_marts', 'fct_orders') }}.order_id
GROUP BY 1

Model Contracts

Model contracts enforce data shape guarantees at the boundary between projects, preventing downstream breakage.

# In the upstream project
models:
  - name: fct_orders
    description: Order-level transaction mart.
    config:
      materialized: table
      contract:
        enforced: true
    constraints:
      - type: primary_key
        columns: [order_id]
      - type: foreign_key
        columns: [customer_id]
        expression: ref('dim_customers')
        columns: [customer_id]
    columns:
      - name: order_id
        data_type: integer
        constraints:
          - type: not_null
      - name: customer_id
        data_type: integer
        constraints:
          - type: not_null
      - name: order_date
        data_type: date
        constraints:
          - type: not_null
      - name: total_amount
        data_type: decimal(18,2)
      - name: status
        data_type: varchar
      - name: updated_at
        data_type: timestamp

When contract: {enforced: true} is set, dbt will:

1. Check that the model's actual columns match the contract definition
2. Add on_schema_change: fail behavior for incremental models
3. Fail the build if any constraint violations are detected

Access Controls (Groups)

# dbt_project.yml — group definitions
groups:
  - name: finance
    owner:
      name: Finance Analytics Team
      email: finance-analytics@company.com
      slack: "#fin-data"
    access: private   # private | protected | public

  - name: marketing
    owner:
      name: Marketing Analytics Team
      email: marketing-analytics@company.com
      slack: "#mkt-data"
    access: private

# In model YAML — assign models to groups
models:
  - name: fct_orders
    group: finance
    access: protected   # Available to other groups via cross-project refs
    config:
      contract:
        enforced: true

  - name: fct_revenue_forecast
    group: finance
    access: private     # Finance team only

  - name: dim_customers
    group: marketing
    access: public      # Available to all projects

Model Versions

When breaking changes are needed, use model versions to provide a migration path:

models:
  - name: fct_orders
    latest_version: 2
    versions:
      - v: 2
        columns:
          - name: order_id
            data_type: integer
          - name: customer_id
            data_type: integer
          - name: total_amount
            data_type: decimal(18,2)
          - include: '*'
            exclude: [legacy_status]
          - name: status_code
            data_type: varchar

      - v: 1
        columns:
          - include: '*'
          - name: legacy_status
            data_type: varchar
        deprecation_date: 2026-12-31

Downstream projects reference specific versions:

SELECT * FROM {{ ref('finance_marts', 'fct_orders', version=2) }}

Governance Best Practices

1. Start with a single project — Only split into Mesh when you have clear ownership boundaries
2. Contracts on public/protected models — Every model with access: public or access: protected should have an enforced contract
3. Semantic versioning — Use model versions (v: 1, v: 2) with deprecation dates for breaking changes
4. dbt Mesh gateway — Use dbt Explorer or dbt Cloud's discovery API for cross-project lineage
5. Shared macros package — Publish common macros (date spine, surrogate keys, cross-db utilities) as a shared dbt package
6. CI/CD for contracts — Run dbt build --select state:modified+contract in CI to validate contract changes

Level 5: Platform & Operations

CLI Command Reference

# ─── Core Commands ───
dbt run                          # Build models
dbt run --select tag:finance     # Build tagged models
dbt run --select +fct_orders     # Build upstream + model
dbt run --select fct_orders+     # Build model + downstream
dbt run --select 1_fct_orders+2  # Build model + 2 levels downstream
dbt run --exclude tag:pii        # Exclude tagged models

# ─── Testing ───
dbt test                         # Run all tests
dbt test --select fct_orders     # Tests for specific model
dbt test --select source:*       # Source freshness only
dbt build                        # run + test (preferred in CI)

# ─── Compilation ───
dbt compile                      # Compile all models, check syntax
dbt compile --no-populate-cache  # Fresh compile
dbt parse                        # Parse project structure (fast)

# ─── Documentation ───
dbt docs generate                # Generate docs site
dbt docs serve                   # Serve docs at localhost:8080

# ─── Debugging ───
dbt debug                        # Check connection + config
dbt debug --config-dir           # Show config directory

# ─── Seeds & Snapshots ───
dbt seed                         # Load CSV seeds
dbt seed --select country_codes  # Specific seed
dbt snapshot                     # Run snapshots

# ─── State & Artifacts ───
dbt ls                           # List all resources in DAG
dbt ls --select source:*         # List sources
dbt ls --resource-type model     # List models
dbt ls --output json             # JSON output for automation

# ─── Freshness ───
dbt source freshness             # Check all source freshness
dbt source freshness --select source:stripe.payments

# ─── Graph Operators ───
# +model     : model and all upstream parents
# model+     : model and all downstream children
# @model     : model and all parents + children
# 2+model    : model and 2 levels of upstream
# model+2    : model and 2 levels of downstream
# tag:name   : all resources with tag
# source:*   : all sources
# fqn:path   : fully qualified name match

Performance Optimization

Query Performance

-- ✅ DO: Use ephemeral for chain-of-transformations
{{
    config(materialized = 'ephemeral')
}}

-- ✅ DO: Filter early in incremental models
{% if is_incremental() %}
    WHERE event_date > (SELECT MAX(event_date) FROM {{ this }})
{% endif %}

-- ✅ DO: Cluster/partition on large incremental tables
{{
    config(
        materialized = 'incremental',
        unique_key = 'event_id',
        partition_by = {'field': 'event_date', 'data_type': 'date'},
        cluster_by = ['customer_id', 'event_type']
    )
}}

-- ❌ DON'T: Use SELECT * in production models
SELECT * FROM {{ ref('stg_orders') }}

-- ✅ DO: Explicit column selection
SELECT
    order_id,
    customer_id,
    order_date,
    status,
    total_amount
FROM {{ ref('stg_orders') }}

-- ❌ DON'T: Join before filtering
SELECT * FROM huge_orders o
JOIN huge_customers c ON o.customer_id = c.id
WHERE o.order_date >= '2024-01-01'

-- ✅ DO: Filter inline in CTEs before joining
WITH recent_orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    WHERE order_date >= '2024-01-01'
),
active_customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
    WHERE is_active = TRUE
)
SELECT * FROM recent_orders o
JOIN active_customers c ON o.customer_id = c.id

Warehouse Cost Optimization

# dbt_project.yml — avoid full refreshes
models:
  my_project:
    marts:
      +full_refresh: false      # Prevent accidental full refreshes
      +on_schema_change: sync_all_columns  # Schema evolution without rebuild

Node Selection for CI Efficiency

# In CI — build only what changed
dbt build --select state:modified+ \
  --defer \
  --state ./target

# Run only new/modified models + their first downstream
dbt build --select state:modified+1

# Run modified models with a specific config
dbt build --select state:modified,config.materialized:incremental

Job Troubleshooting Guide

Common Errors

Debugging Workflow

1. dbt parse              → Validate syntax and structure first
2. dbt compile            → Check SQL compilation
3. dbt run --select model → Isolate the failing model
4. Check logs for SQL     → Inspect the compiled SQL query
5. Test upstream models   → Verify data quality at source
6. dbt --debug run        → Full debug mode with verbose logging

dbt MCP Server Configuration

For AI-assisted dbt development, configure the dbt MCP server:

// .cursor/mcp.json or claude_desktop_config.json
{
  "mcpServers": {
    "dbt": {
      "command": "uvx",
      "args": ["dbt-mcp-server"],
      "env": {
        "DBT_PROFILES_DIR": "/Users/username/.dbt",
        "DBT_PROJECT_DIR": "/Users/username/projects/my_dbt_project",
        "DBT_TARGET": "dev"
      }
    }
  }
}

dbt MCP Server capabilities:

• dbt_compile: Compile a model and return compiled SQL
• dbt_run: Execute model builds
• dbt_test: Run tests for specified models
• dbt_docs_generate: Generate project documentation
• dbt_list: List project resources and DAG
• dbt_semantic_query: Execute MetricFlow queries

Migration Guide (Older → Modern dbt)

Key Migration Steps (v1.5+ → v1.9+)

# Upgrade dbt-core and adapters
pip install --upgrade dbt-core dbt-snowflake  # or dbt-bigquery, dbt-postgres, etc.

# Reinstall packages
dbt deps

# Recompile entire project
dbt compile --no-populate-cache

# Run tests to validate
dbt test --select source:*
dbt build --select state:modified+

Cross-Domain Best Practices

YAML DRY Patterns (Doc Blocks)

# macros/doc_blocks.yml
version: 2

# Reusable column descriptions
docs:
  - name: pk_column
    description: Primary key. Unique identifier for this reco

…(truncated)