Dbt Data Transformation

dbt Data Transformation

A comprehensive skill for mastering dbt (data build tool) for analytics engineering. This skill covers model development, testing strategies, documentation practices, incremental builds, Jinja templating, macro development, package management, and production deployment workflows.

When to Use This Skill

Use this skill when:

• Building data transformation pipelines for analytics and business intelligence
• Creating a data warehouse with modular, testable SQL transformations
• Implementing ELT (Extract, Load, Transform) workflows
• Developing dimensional models (facts, dimensions) for analytics
• Managing complex SQL dependencies and data lineage
• Creating reusable data transformation logic across projects
• Testing data quality and implementing data contracts
• Documenting data models and business logic
• Building incremental models for large datasets
• Orchestrating dbt with tools like Airflow, Dagster, or dbt Cloud
• Migrating legacy ETL processes to modern ELT architecture
• Implementing DataOps practices for analytics teams

Core Concepts

What is dbt?

dbt (data build tool) enables analytics engineers to transform data in their warehouse more effectively. It's a development framework that brings software engineering best practices to data transformation:

• Version Control: SQL transformations as code in Git
• Testing: Built-in data quality testing framework
• Documentation: Auto-generated, searchable data dictionary
• Modularity: Reusable SQL through refs and macros
• Lineage: Automatic dependency resolution and visualization
• Deployment: CI/CD for data transformations

The dbt Workflow

1. Develop: Write SQL SELECT statements as models
2. Test: Define data quality tests
3. Document: Add descriptions and metadata
4. Build: dbt run compiles and executes models
5. Test: dbt test validates data quality
6. Deploy: CI/CD pipelines deploy to production

Key dbt Entities

1. Models: SQL SELECT statements that define data transformations 2. Sources: Raw data tables in your warehouse 3. Seeds: CSV files loaded into your warehouse 4. Tests: Data quality assertions 5. Macros: Reusable Jinja-SQL functions 6. Snapshots: Type 2 slowly changing dimension captures 7. Exposures: Downstream uses of dbt models (dashboards, ML models) 8. Metrics: Business metric definitions

Model Development

Basic Model Structure

A dbt model is a SELECT statement saved as a .sql file:

-- models/staging/stg_orders.sql

with source as (
    select * from {{ source('jaffle_shop', 'orders') }}
),

renamed as (
    select
        id as order_id,
        user_id as customer_id,
        order_date,
        status,
        _etl_loaded_at
    from source
)

select * from renamed

Key Points:

• Models are SELECT statements only (no DDL)
• Use CTEs (Common Table Expressions) for readability
• Reference sources with {{ source() }}
• dbt handles CREATE/INSERT logic based on materialization

The ref() Function

Reference other models using {{ ref() }}:

-- models/marts/fct_orders.sql

with orders as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select * from {{ ref('stg_customers') }}
),

joined as (
    select
        orders.order_id,
        orders.order_date,
        customers.customer_name,
        orders.status
    from orders
    left join customers
        on orders.customer_id = customers.customer_id
)

select * from joined

Benefits of ref():

• Builds dependency graph automatically
• Resolves to correct schema/database
• Enables testing in dev without affecting prod
• Powers lineage visualization

The source() Function

Define and reference raw data sources:

# models/staging/sources.yml

version: 2

sources:
  - name: jaffle_shop
    description: Raw data from the Jaffle Shop application
    database: raw
    schema: jaffle_shop
    tables:
      - name: orders
        description: One record per order
        columns:
          - name: id
            description: Primary key for orders
            tests:
              - unique
              - not_null
          - name: user_id
            description: Foreign key to customers
          - name: order_date
            description: Date order was placed
          - name: status
            description: Order status (completed, pending, cancelled)

-- Reference the source
select * from {{ source('jaffle_shop', 'orders') }}

Source Features:

• Document raw data tables
• Test source data quality
• Track freshness with freshness config
• Separate source definitions from transformations

Model Organization

Recommended project structure:

models/
├── staging/           # One-to-one with source tables
│   ├── jaffle_shop/
│   │   ├── _jaffle_shop__sources.yml
│   │   ├── _jaffle_shop__models.yml
│   │   ├── stg_jaffle_shop__orders.sql
│   │   └── stg_jaffle_shop__customers.sql
│   └── stripe/
│       ├── _stripe__sources.yml
│       ├── _stripe__models.yml
│       └── stg_stripe__payments.sql
├── intermediate/      # Purpose-built transformations
│   └── int_orders_joined.sql
└── marts/            # Business-defined entities
    ├── core/
    │   ├── _core__models.yml
    │   ├── dim_customers.sql
    │   └── fct_orders.sql
    └── marketing/
        └── fct_customer_sessions.sql

Naming Conventions:

• stg_: Staging models (one-to-one with sources)
• int_: Intermediate models (not exposed to end users)
• fct_: Fact tables
• dim_: Dimension tables

Materializations

Materializations determine how dbt builds models in your warehouse:

1. View (Default)

{{ config(materialized='view') }}

select * from {{ ref('base_model') }}

Characteristics:

• Lightweight, no data stored
• Query runs each time view is accessed
• Best for: Small datasets, models queried infrequently
• Fast to build, slower to query

2. Table

{{ config(materialized='table') }}

select * from {{ ref('base_model') }}

Characteristics:

• Full table rebuild on each run
• Data physically stored
• Best for: Small to medium datasets, heavily queried models
• Slower to build, faster to query

3. Incremental

{{ config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='fail'
) }}

select * from {{ source('jaffle_shop', 'orders') }}

{% if is_incremental() %}
    -- Only process new/updated records
    where order_date > (select max(order_date) from {{ this }})
{% endif %}

Characteristics:

• Only processes new data on subsequent runs
• First run builds full table
• Best for: Large datasets, event/time-series data
• Fast incremental builds, maintains historical data

Incremental Strategies:

-- Append (default): Add new rows only
{{ config(
    materialized='incremental',
    incremental_strategy='append'
) }}

-- Merge: Upsert based on unique_key
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

-- Delete+Insert: Delete matching records, insert new
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='delete+insert'
) }}

4. Ephemeral

{{ config(materialized='ephemeral') }}

select * from {{ ref('base_model') }}

Characteristics:

• Not built in warehouse
• Interpolated as CTE in dependent models
• Best for: Lightweight transformations, avoiding view proliferation
• No storage, compiled into downstream models

Configuration Comparison

*After initial full build

Testing

Schema Tests

Built-in generic tests defined in YAML:

# models/staging/stg_orders.yml

version: 2

models:
  - name: stg_orders
    description: Staged order data
    columns:
      - name: order_id
        description: Primary key
        tests:
          - unique
          - not_null

      - name: customer_id
        description: Foreign key to customers
        tests:
          - not_null
          - relationships:
              to: ref('stg_customers')
              field: customer_id

      - name: status
        description: Order status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned', 'cancelled']

      - name: order_total
        description: Total order amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

Built-in Tests:

• unique: No duplicate values
• not_null: No null values
• accepted_values: Value in specified list
• relationships: Foreign key validation

Custom Data Tests

Create custom tests in tests/ directory:

-- tests/assert_positive_order_totals.sql

select
    order_id,
    order_total
from {{ ref('fct_orders') }}
where order_total < 0

How it works:

• Test fails if query returns any rows
• Query should return failing records
• Can use any SQL logic

Advanced Testing Patterns

-- Test for data freshness
-- tests/assert_orders_are_fresh.sql

with latest_order as (
    select max(order_date) as max_date
    from {{ ref('fct_orders') }}
)

select max_date
from latest_order
where max_date < current_date - interval '1 day'

-- Test for referential integrity across time
-- tests/assert_no_orphaned_orders.sql

select
    o.order_id,
    o.customer_id
from {{ ref('fct_orders') }} o
left join {{ ref('dim_customers') }} c
    on o.customer_id = c.customer_id
where c.customer_id is null

Testing with dbt_utils

# Requires dbt-utils package

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          # Test for uniqueness across multiple columns
          - dbt_utils.unique_combination_of_columns:
              combination_of_columns:
                - order_id
                - order_date

          # Test for sequential values
          - dbt_utils.sequential_values:
              interval: 1

          # Test that values match regex
          - dbt_utils.not_null_proportion:
              at_least: 0.95

Test Severity Levels

models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique:
              severity: error  # Fail build (default)
          - not_null:
              severity: warn   # Warning only

Documentation

Model Documentation

# models/marts/core/_core__models.yml

version: 2

models:
  - name: fct_orders
    description: |
      Order fact table containing one row per order with associated
      customer and payment information. This is the primary table for
      order analytics and reporting.

      **Grain:** One row per order

      **Refresh:** Incremental, updates daily at 2 AM UTC

      **Notes:**
      - Includes cancelled orders (filter with status column)
      - Payment info joined from Stripe data
      - Customer info joined from application database

    columns:
      - name: order_id
        description: Primary key for orders table
        tests:
          - unique
          - not_null

      - name: customer_id
        description: |
          Foreign key to dim_customers. Links to customer who placed the order.
          **Note:** May be null for guest checkout orders.

      - name: order_date
        description: Date order was placed (UTC timezone)

      - name: status
        description: |
          Current order status. Possible values:
          - `placed`: Order received, not yet processed
          - `shipped`: Order shipped to customer
          - `completed`: Order delivered and confirmed
          - `returned`: Order returned by customer
          - `cancelled`: Order cancelled before shipment

      - name: order_total
        description: Total order amount in USD including tax and shipping

Documentation Blocks

Create reusable documentation:

<!-- models/docs.md -->

{% docs order_status %}

Order status indicates the current state of an order in our fulfillment pipeline.

| Status | Description | Next Steps |
|--------|-------------|------------|
| placed | Order received | Inventory check |
| shipped | En route to customer | Track shipment |
| completed | Delivered successfully | Request feedback |
| returned | Customer return initiated | Process refund |
| cancelled | Order cancelled | Update inventory |

{% enddocs %}

{% docs customer_id %}

Unique identifier for customers. This ID is:
- Generated by the application on account creation
- Persistent across orders
- Used to track customer lifetime value
- **Note:** NULL for guest checkouts

{% enddocs %}

Reference documentation blocks:

models:
  - name: fct_orders
    columns:
      - name: status
        description: "{{ doc('order_status') }}"

      - name: customer_id
        description: "{{ doc('customer_id') }}"

Generating Documentation

# Generate documentation site
dbt docs generate

# Serve documentation locally
dbt docs serve --port 8001

# View in browser at http://localhost:8001

Documentation Features:

• Interactive lineage graph (DAG visualization)
• Searchable model catalog
• Column-level documentation
• Source freshness tracking
• Test coverage visibility
• Compiled SQL preview

Documentation Best Practices

1. Document at all levels: Project, models, columns, sources 2. Explain business logic: Why transformations exist 3. Define grain explicitly: One row represents... 4. Note refresh schedules: How often data updates 5. Document assumptions: Edge cases, known issues 6. Link to external resources: Confluence, wiki, dashboards

Incremental Models

Basic Incremental Pattern

{{ config(
    materialized='incremental',
    unique_key='event_id'
) }}

with source as (
    select
        event_id,
        user_id,
        event_timestamp,
        event_type,
        event_properties
    from {{ source('analytics', 'events') }}

    {% if is_incremental() %}
        -- Only process events newer than existing data
        where event_timestamp > (select max(event_timestamp) from {{ this }})
    {% endif %}
)

select * from source

Key Components:

• is_incremental(): True after first run
• {{ this }}: References current model's table
• unique_key: Column(s) for deduplication

Incremental with Merge Strategy

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    merge_update_columns=['status', 'updated_at'],
    merge_exclude_columns=['created_at']
) }}

with orders as (
    select
        order_id,
        customer_id,
        order_date,
        status,
        order_total,
        current_timestamp() as updated_at,
        case
            when status = 'placed' then current_timestamp()
            else null
        end as created_at
    from {{ source('ecommerce', 'orders') }}

    {% if is_incremental() %}
        -- Look back 3 days to catch late-arriving updates
        where order_date >= (select max(order_date) - interval '3 days' from {{ this }})
    {% endif %}
)

select * from orders

Merge Strategy Features:

• Updates existing records based on unique_key
• Inserts new records
• Optional: Specify which columns to update/exclude
• Best for: Slowly changing data, updates to historical records

Incremental with Delete+Insert

{{ config(
    materialized='incremental',
    unique_key=['date', 'customer_id'],
    incremental_strategy='delete+insert'
) }}

with daily_metrics as (
    select
        date_trunc('day', order_timestamp) as date,
        customer_id,
        count(*) as order_count,
        sum(order_total) as total_revenue
    from {{ ref('fct_orders') }}

    {% if is_incremental() %}
        where date_trunc('day', order_timestamp) >= (
            select max(date) - interval '7 days' from {{ this }}
        )
    {% endif %}

    group by 1, 2
)

select * from daily_metrics

Delete+Insert Strategy:

• Deletes all rows matching unique_key
• Inserts new rows
• Best for: Aggregated data, full partition replacement
• More efficient than merge for bulk updates

Handling Late-Arriving Data

{{ config(
    materialized='incremental',
    unique_key='order_id'
) }}

select
    order_id,
    customer_id,
    order_date,
    status,
    _loaded_at
from {{ source('ecommerce', 'orders') }}

{% if is_incremental() %}
    -- Use _loaded_at instead of order_date to catch updates
    where _loaded_at > (select max(_loaded_at) from {{ this }})

    -- OR use a lookback window
    -- where order_date > (select max(order_date) - interval '3 days' from {{ this }})
{% endif %}

Incremental with Partitioning

{{ config(
    materialized='incremental',
    unique_key='event_id',
    partition_by={
        'field': 'event_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['user_id', 'event_type']
) }}

select
    event_id,
    user_id,
    event_type,
    event_timestamp,
    date(event_timestamp) as event_date
from {{ source('analytics', 'raw_events') }}

{% if is_incremental() %}
    where date(event_timestamp) > (select max(event_date) from {{ this }})
{% endif %}

Partition Benefits:

• Improved query performance
• Cost optimization (scan less data)
• Efficient incremental processing
• Better for time-series data

Full Refresh Capability

# Force full rebuild of incremental models
dbt run --full-refresh

# Full refresh specific model
dbt run --select my_incremental_model --full-refresh

Macros & Jinja

Basic Macro Structure

-- macros/cents_to_dollars.sql

{% macro cents_to_dollars(column_name, precision=2) %}
    round({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}

Usage:

select
    order_id,
    {{ cents_to_dollars('amount_cents') }} as amount_dollars
from {{ ref('stg_orders') }}

Reusable Data Quality Macros

-- macros/test_not_negative.sql

{% macro test_not_negative(model, column_name) %}

select
    {{ column_name }}
from {{ model }}
where {{ column_name }} < 0

{% endmacro %}

Date Spine Macro

-- macros/date_spine.sql

{% macro date_spine(start_date, end_date) %}

with date_spine as (
    {{ dbt_utils.date_spine(
        datepart="day",
        start_date="cast('" ~ start_date ~ "' as date)",
        end_date="cast('" ~ end_date ~ "' as date)"
    ) }}
)

select
    date_day
from date_spine

{% endmacro %}

Dynamic SQL Generation

-- macros/pivot_metric.sql

{% macro pivot_metric(metric_column, group_by_column, values) %}

select
    {{ group_by_column }},
    {% for value in values %}
        sum(case when status = '{{ value }}' then {{ metric_column }} else 0 end)
            as {{ value }}_{{ metric_column }}
        {% if not loop.last %},{% endif %}
    {% endfor %}
from {{ ref('fct_orders') }}
group by 1

{% endmacro %}

Usage:

{{ pivot_metric(
    metric_column='order_total',
    group_by_column='customer_id',
    values=['completed', 'pending', 'cancelled']
) }}

Grant Permissions Macro

-- macros/grant_select.sql

{% macro grant_select(schema, role) %}

{% set sql %}
    grant select on all tables in schema {{ schema }} to {{ role }};
{% endset %}

{% do run_query(sql) %}
{% do log("Granted select on " ~ schema ~ " to " ~ role, info=True) %}

{% endmacro %}

Usage in hooks:

# dbt_project.yml

on-run-end:
  - "{{ grant_select(target.schema, 'analyst_role') }}"

Environment-Specific Logic

-- macros/generate_schema_name.sql

{% macro generate_schema_name(custom_schema_name, node) -%}

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

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

    {%- else -%}
        {{ default_schema }}_{{ custom_schema_name | trim }}
    {%- endif -%}

{%- endmacro %}

Audit Column Macro

-- macros/add_audit_columns.sql

{% macro add_audit_columns() %}

    current_timestamp() as dbt_updated_at,
    current_timestamp() as dbt_created_at,
    '{{ var("dbt_user") }}' as dbt_updated_by

{% endmacro %}

Usage:

select
    order_id,
    customer_id,
    order_total,
    {{ add_audit_columns() }}
from {{ ref('stg_orders') }}

Jinja Control Structures

-- Conditionals
{% if target.name == 'prod' %}
    from {{ source('production', 'orders') }}
{% else %}
    from {{ source('development', 'orders') }}
{% endif %}

-- Loops
{% for status in ['placed', 'shipped', 'completed'] %}
    sum(case when status = '{{ status }}' then 1 else 0 end) as {{ status }}_count
    {% if not loop.last %},{% endif %}
{% endfor %}

-- Set variables
{% set payment_methods = ['credit_card', 'paypal', 'bank_transfer'] %}

{% for method in payment_methods %}
    count(distinct case when payment_method = '{{ method }}'
        then customer_id end) as {{ method }}_customers
    {% if not loop.last %},{% endif %}
{% endfor %}

Package Management

Installing Packages

# packages.yml

packages:
  # dbt-utils: Essential utility macros
  - package: dbt-labs/dbt_utils
    version: 1.1.1

  # Audit helper: Compare datasets
  - package: dbt-labs/audit_helper
    version: 0.9.0

  # Codegen: Code generation utilities
  - package: dbt-labs/codegen
    version: 0.11.0

  # Custom package from Git
  - git: "https://github.com/your-org/dbt-custom-package.git"
    revision: main

  # Local package
  - local: ../dbt-shared-macros

Install packages:

dbt deps

Using dbt_utils

-- Surrogate key generation
select
    {{ dbt_utils.generate_surrogate_key(['order_id', 'line_item_id']) }} as order_line_id,
    order_id,
    line_item_id
from {{ ref('stg_order_lines') }}

-- Union multiple tables
{{ dbt_utils.union_relations(
    relations=[
        ref('orders_2022'),
        ref('orders_2023'),
        ref('orders_2024')
    ]
) }}

-- Get column values as list
{% set statuses = dbt_utils.get_column_values(
    table=ref('stg_orders'),
    column='status'
) %}

-- Pivot table
{{ dbt_utils.pivot(
    column='metric_name',
    values=dbt_utils.get_column_values(table=ref('metrics'), column='metric_name'),
    agg='sum',
    then_value='metric_value',
    else_value=0,
    prefix='',
    suffix='_total'
) }}

Creating Custom Packages

Project structure for a package:

dbt-custom-package/
├── dbt_project.yml
├── macros/
│   ├── custom_test.sql
│   └── custom_macro.sql
├── models/
│   └── example_model.sql
└── README.md

# dbt_project.yml for custom package

name: 'custom_package'
version: '1.0.0'
config-version: 2

require-dbt-version: [">=1.0.0", "<2.0.0"]

Package Versioning

# Semantic versioning
packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.0.0", "<2.0.0"]  # Any 1.x version

  # Exact version
  - package: dbt-labs/dbt_utils
    version: 1.1.1

  # Git branch/tag
  - git: "https://github.com/org/package.git"
    revision: v1.2.3

  # Latest from branch
  - git: "https://github.com/org/package.git"
    revision: main

Production Workflows

CI/CD Pipeline (GitHub Actions)

# .github/workflows/dbt_ci.yml

name: dbt CI

on:
  pull_request:
    branches: [main]

jobs:
  dbt_run:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dbt
        run: |
          pip install dbt-core dbt-snowflake

      - name: Install dbt packages
        run: dbt deps

      - name: Run dbt debug
        run: dbt debug
        env:
          DBT_SNOWFLAKE_ACCOUNT: ${{ secrets.DBT_SNOWFLAKE_ACCOUNT }}
          DBT_SNOWFLAKE_USER: ${{ secrets.DBT_SNOWFLAKE_USER }}
          DBT_SNOWFLAKE_PASSWORD: ${{ secrets.DBT_SNOWFLAKE_PASSWORD }}

      - name: Run dbt models (modified only)
        run: dbt run --select state:modified+ --state ./prod_manifest
        env:
          DBT_SNOWFLAKE_ACCOUNT: ${{ secrets.DBT_SNOWFLAKE_ACCOUNT }}
          DBT_SNOWFLAKE_USER: ${{ secrets.DBT_SNOWFLAKE_USER }}
          DBT_SNOWFLAKE_PASSWORD: ${{ secrets.DBT_SNOWFLAKE_PASSWORD }}

      - name: Run dbt tests
        run: dbt test --select state:modified+
        env:
          DBT_SNOWFLAKE_ACCOUNT: ${{ secrets.DBT_SNOWFLAKE_ACCOUNT }}
          DBT_SNOWFLAKE_USER: ${{ secrets.DBT_SNOWFLAKE_USER }}
          DBT_SNOWFLAKE_PASSWORD: ${{ secrets.DBT_SNOWFLAKE_PASSWORD }}

Slim CI (Test Changed Models Only)

# Store production manifest
dbt compile --target prod
cp target/manifest.json ./prod_manifest/

# In CI: Test only changed models and downstream dependencies
dbt test --select state:modified+ --state ./prod_manifest

Production Deployment

# .github/workflows/dbt_prod.yml

name: dbt Production Deploy

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dbt
        run: pip install dbt-core dbt-snowflake

      - name: Install packages
        run: dbt deps

      - name: Run dbt seed
        run: dbt seed --target prod

      - name: Run dbt run
        run: dbt run --target prod

      - name: Run dbt test
        run: dbt test --target prod

      - name: Generate docs
        run: dbt docs generate --target prod

      - name: Upload docs to S3
        run: |
          aws s3 sync target/ s3://dbt-docs-bucket/
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

Orchestration with Airflow

# dags/dbt_dag.py

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'analytics',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'dbt_production',
    default_args=default_args,
    description='Run dbt models in production',
    schedule_interval='0 2 * * *',  # 2 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['dbt', 'analytics'],
) as dag:

    dbt_deps = BashOperator(
        task_id='dbt_deps',
        bash_command='cd /opt/dbt && dbt deps',
    )

    dbt_seed = BashOperator(
        task_id='dbt_seed',
        bash_command='cd /opt/dbt && dbt seed --target prod',
    )

    dbt_run_staging = BashOperator(
        task_id='dbt_run_staging',
        bash_command='cd /opt/dbt && dbt run --select staging.* --target prod',
    )

    dbt_run_marts = BashOperator(
        task_id='dbt_run_marts',
        bash_command='cd /opt/dbt && dbt run --select marts.* --target prod',
    )

    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /opt/dbt && dbt test --target prod',
    )

    dbt_docs = BashOperator(
        task_id='dbt_docs',
        bash_command='cd /opt/dbt && dbt docs generate --target prod',
    )

    # Define task dependencies
    dbt_deps >> dbt_seed >> dbt_run_staging >> dbt_run_marts >> dbt_test >> dbt_docs

dbt Cloud Integration

# dbt_cloud.yml

# Environment configuration
environments:
  - name: Production
    dbt_version: 1.7.latest
    type: deployment

  - name: Development
    dbt_version: 1.7.latest
    type: development

# Job configuration
jobs:
  - name: Production Run
    environment: Production
    triggers:
      schedule:
        cron: "0 2 * * *"  # 2 AM daily
    commands:
      - dbt deps
      - dbt seed
      - dbt run
      - dbt test

  - name: CI Check
    environment: Development
    triggers:
      github_webhook: true
    commands:
      - dbt deps
      - dbt run --select state:modified+
      - dbt test --select state:modified+

Monitoring & Alerting

-- macros/post_hook_monitoring.sql

{% macro monitor_row_count(threshold=0) %}

{% if execute %}
    {% set row_count_query %}
        select count(*) as row_count from {{ this }}
    {% endset %}

    {% set results = run_query(row_count_query) %}
    {% set row_count = results.columns[0].values()[0] %}

    {% if row_count < threshold %}
        {{ exceptions.raise_compiler_error("Row count " ~ row_count ~ " below threshold " ~ threshold) }}
    {% endif %}

    {{ log("Model " ~ this ~ " has " ~ row_count ~ " rows", info=True) }}
{% endif %}

{% endmacro %}

Usage:

{{ config(
    post_hook="{{ monitor_row_count(threshold=1000) }}"
) }}

select * from {{ ref('stg_orders') }}

Best Practices

Naming Conventions

Models:

stg_[source]__[entity].sql        # Staging: stg_stripe__payments.sql
int_[entity]_[verb].sql            # Intermediate: int_orders_joined.sql
fct_[entity].sql                   # Fact: fct_orders.sql
dim_[entity].sql                   # Dimension: dim_customers.sql

Tests:

assert_[description].sql           # assert_positive_order_totals.sql

Macros:

[verb]_[noun].sql                  # generate_surrogate_key.sql

SQL Style Guide

-- ✓ Good: Clear CTEs, proper formatting
with orders as (
    select
        order_id,
        customer_id,
        order_date,
        status
    from {{ ref('stg_orders') }}
    where status != 'cancelled'
),

customers as (
    select
        customer_id,
        customer_name,
        customer_email
    from {{ ref('dim_customers') }}
),

final as (
    select
        orders.order_id,
        orders.order_date,
        customers.customer_name,
        orders.status
    from orders
    left join customers
        on orders.customer_id = customers.customer_id
)

select * from final

-- ✗ Bad: Nested subqueries, poor formatting
select o.order_id, o.order_date, c.customer_name, o.status from (
select order_id, customer_id, order_date, status from {{ ref('stg_orders') }}
where status != 'cancelled') o left join (select customer_id, customer_name from
{{ ref('dim_customers') }}) c on o.customer_id = c.customer_id

Performance Optimization

1. Use Incremental Models for Large Tables

-- Process only new data
{{ config(materialized='incremental') }}

select * from {{ source('events', 'page_views') }}
{% if is_incremental() %}
    where event_timestamp > (select max(event_timestamp) from {{ this }})
{% endif %}

2. Leverage Clustering and Partitioning

{{ config(
    materialized='table',
    partition_by={'field': 'order_date', 'data_type': 'date'},
    cluster_by=['customer_id', 'status']
) }}

3. Reduce Data Scanned

-- ✓ Good: Filter early
with source as (
    select *
    from {{ source('app', 'events') }}
    where event_date >= '2024-01-01'  -- Filter in source CTE
)

-- ✗ Bad: Filter late
with source as (
    select * from {{ source('app', 'events') }}
)

select * from source
where event_date >= '2024-01-01'  -- Filtering after full scan

4. Use Ephemeral for Simple Transformations

-- Avoid creating unnecessary views
{{ config(materialized='ephemeral') }}

select
    order_id,
    lower(trim(status)) as status_clean
from {{ ref('stg_orders') }}

Project Structure Best Practices

1. Layer Your Transformations

Staging → Intermediate → Marts
  ↓           ↓            ↓
 1:1      Purpose-built  Business
Sources    Logic        Entities

2. Modularize Complex Logic

-- Instead of one massive model, break it down:

-- intermediate/int_order_items_aggregated.sql
-- intermediate/int_customer_lifetime_value.sql
-- intermediate/int_payment_summaries.sql

-- marts/fct_orders.sql (combines intermediate models)

3. Use Consistent File Organization

models/
├── staging/
│   └── [source]/
│       ├── _[source]__sources.yml
│       ├── _[source]__models.yml
│       └── stg_[source]__[table].sql
├── intermediate/
│   └── int_[purpose].sql
└── marts/
    └── [business_area]/
        ├── _[area]__models.yml
        └── [model_type]_[entity].sql

Testing Strategy

1. Test at Multiple Levels

# Source tests: Data quality at ingestion
sources:
  - name: raw_data
    tables:
      - name: orders
        columns:
          - name: id
            tests: [unique, not_null]

# Model tests: Transformation logic
models:
  - name: fct_orders
    tests:
      - dbt_utils.expression_is_true:
          expression: "order_total >= 0"
    columns:
      - name: order_id
        tests: [unique, not_null]

# Custom tests: Business logic
# tests/assert_revenue_reconciliation.sql

2. Use Appropriate Test Severity

# Critical tests: error (fail build)
# Nice-to-have tests: warn (log but don't fail)

tests:
  - unique:
      severity: error
  - dbt_utils.not_null_proportion:
      at_least: 0.95
      severity: warn

3. Test Coverage Goals

• 100% of primary keys: unique + not_null
• 100% of foreign keys: relationships tests
• All business logic: custom data tests
• Critical calculations: expression tests

Documentation Standards

1. Document Every Model

models:
  - name: fct_orders
    description: |
      **Purpose:** [Why this model exists]
      **Grain:** [One row represents...]
      **Refresh:** [When and how often]
      **Consumers:** [Who uses this]

2. Document Complex Logic

-- Use comments for complex business rules
select
    order_id,
    -- Revenue recognition: Only count completed orders
    -- cancelled within 30 days (per finance policy 2024-03)
    case
        when status = 'completed'
            and datediff('day', order_date, current_date) > 30
        then order_total
        else 0
    end as recognized_revenue
from {{ ref('stg_orders') }}

3. Keep Docs Updated

• Update docs when logic changes
• Review docs during code reviews
• Generate docs regularly: dbt docs generate

20 Detailed Examples

Example 1: Basic Staging Model

-- models/staging/jaffle_shop/stg_jaffle_shop__customers.sql

with source as (
    select * from {{ source('jaffle_shop', 'customers') }}
),

renamed as (
    select
        id as customer_id,
        first_name,
        last_name,
        first_name || ' ' || last_name as customer_name,
        email,
        _loaded_at
    from source
)

select * from renamed

Example 2: Fact Table with Multiple Joins

-- models/marts/core/fct_orders.sql

{{ config(
    materialized='table',
    tags=['core', 'daily']
) }}

with orders as (
    select * from {{ ref('stg_jaffle_shop__orders') }}
),

customers as (
    select * from {{ ref('dim_customers') }}
),

payments as (
    select
        order_id,
        sum(amount) as total_payment_amount
    from {{ ref('stg_stripe__payments') }}
    where status = 'success'
    group by 1
),

final as (
    select
        orders.order_id,
        orders.customer_id,
        customers.customer_name,
        orders.order_date,
        orders.status,
        coalesce(payments.total_payment_amount, 0) as order_total,
        {{ add_audit_columns() }}
    from orders
    left join customers
        on orders.customer_id = customers.customer_id
    left join payments
        on orders.order_id = payments.order_id
)

select * from final

Example 3: Incremental Event Table

-- models/marts/analytics/fct_page_views.sql

{{ config(
    materialized='incremental',
    unique_key='page_view_id',
    partition_by={
        'field': 'event_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['user_id', 'page_path']
) }}

with events as (
    select
        event_id as page_view_id,
        user_id,
        session_id,
        event_timestamp,
        date(event_timestamp) as event_date,
        event_properties:page_path::string as page_path,
        event_properties:referrer::string as referrer,
        _loaded_at
    from {{ source('analytics', 'raw_events') }}
    where event_type = 'page_view'

    {% if is_incremental() %}
        -- Use _loaded_at to catch late-arriving data
        and _loaded_at > (select max(_loaded_at) from {{ this }})
    {% endif %}
),

enriched as (
    select
        page_view_id,
        user_id,
        session_id,
        event_timestamp,
        event_date,
        page_path,
        referrer,
        -- Parse URL components
        split_part(page_path, '?', 1) as page_path_clean,
        case
            when referrer like '%google%' then 'Google'
            when referrer like '%facebook%' then 'Facebook'
            when referrer is null then 'Direct'
            else 'Other'
        end as referrer_source,
        _loaded_at
    from events
)

select * from enriched

Example 4: Customer Dimension with SCD Type 2

-- models/marts/core/dim_customers.sql

{{ config(
    materialized='table',
    unique_key='customer_key'
) }}

with customers as (
    select * from {{ ref('stg_jaffle_shop__customers') }}
),

customer_orders as (
    select
        customer_id,
        min(order_date) as first_order_date,
        max(order_date) as most_recent_order_date,
        count(order_id) as total_orders
    from {{ ref('fct_orders') }}
    group by 1
),

final as (
    select
        {{ dbt_utils.generate_surrogate_key(['customers.customer_id', 'customers._loaded_at']) }}
            as customer_key,
        customers.customer_id,
        customers.customer_name,
        customers.email,
        customer_orders.first_order_date,
        customer_orders.most_recent_order_date,
        customer_orders.total_orders,
        case
            when customer_orders.total_orders >= 10 then 'VIP'
            when customer_orders.total_orders >= 5 then 'Regular'
            when customer_orders.total_orders >= 1 then 'New'
            else 'Prospect'
        end as customer_segment,
        customers._loaded_at as effective_from,
        null as effective_to,
        true as is_current
    from customers
    left join customer_orders
        on customers.customer_id = customer_orders.customer_id
)

select * from final

Example 5: Aggregated Metrics Table

-- models/marts/analytics/daily_order_metrics.sql

{{ config(
    materialized='incremental',
    unique_key=['metric_date', 'status'],
    incremental_strategy='delete+insert'
) }}

with orders as (
    select * from {{ ref('fct_orders') }}

    {% if is_incremental() %}
        where order_date >= (select max(metric_date) - interval '7 days' from {{ this }})
    {% endif %}
),

daily_metrics as (
    select
        date_trunc('day', order_date) as metric_date,
        status,
        count(distinct order_id) as order_count,
        count(distinct customer_id) as unique_customers,
        sum(order_total) as total_revenue,
        avg(order_total) as avg_order_value,
        min(order_total) as min_order_value,
        max(order_total) as max_order_value,
        percentile_cont(0.5) within group (order by order_total) as median_order_value
    from orders
    group by 1, 2
)

select * from daily_metrics

Example 6: Pivoted Metrics Using Macro

-- models/marts/analytics/customer_order_status_summary.sql

with orders as (
    select
        customer_id,
        status,
        order_total
    from {{ ref('fct_orders') }}
)

select
    customer_id,
    {% for status in ['placed', 'shipped', 'completed', 'returned', 'cancelled'] %}
        sum(case when status = '{{ status }}' then 1 else 0 end)
            as {{ status }}_count,
        sum(case when status = '{{ status }}' then order_total else 0 end)
            as {{ status }}_revenue
        {% if not loop.last %},{% endif %}
    {% endfor %}
from orders
group by 1

Example 7: Snapshot for SCD Type 2

-- snapshots/customers_snapshot.sql

{% snapshot customers_snapshot %}

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

select
    customer_id,
    customer_name,
    email,
    customer_segment,
    updated_at
from {{ source('jaffle_shop', 'customers') }}

{% endsnapshot %}

Example 8: Funnel Analysis Model

-- models/marts/analytics/conversion_funnel.sql

with page_views as (
    select
        user_id,
        session_id,
        min(event_timestamp) as session_start
    from {{ ref('fct_page_views') }}
    where event_date >= current_date - interval '30 days'
    group by 1, 2
),

product_views as (
    select distinct
        user_id,
        session_id
    from {{ ref('fct_page_views') }}
    where page_path like '/product/%'
        and event_date >= current_date - interval '30 days'
),

add_to_cart as (
    select distinct
        user_id,
        session_id
    from {{ ref('fct_events') }}
    where event_type = 'add_to_cart'
        and event_date >= current_date - interval '30 days'
),

checkout_started as (
    select distinct
        user_id,
        session_id
    from {{ ref('fct_events') }}
    where event_type = 'checkout_started'
        and event_date >= current_date - interval '30 days'
),

orders as (
    select distinct
        customer_id as user_id,
        session_id
    from {{ ref('fct_orders') }}
    where order_date >= current_date - interval '30 days'
        and status = 'completed'
),

funnel as (
    select
        count(distinct page_views.session_id) as sessions,
        count(distinct product_views.session_id) as product_views,
        count(distinct add_to_cart.session_id) as add_to_cart,
        count(distinct checkout_started.session_id) as checkout_started,
        count(distinct orders.session_id) as completed_orders
    from page_views
    left join product_views using (session_id)
    left join add_to_cart using (session_id)
    left join checkout_started using (session_id)
    left join orders using (session_id)
),

funnel_metrics as (
    select
        sessions,
        product_views,
        round(100.0 * product_views / nullif(sessions, 0), 2) as pct_product_views,
        add_to_cart,
        round(100.0 * add_to_cart / nullif(product_views, 0), 2) as pct_add_to_cart,
        checkout_started,
        round(100.0 * checkout_started / nullif(add_to_cart, 0), 2) as pct_checkout_started,
        completed_orders,
        round(100.0 * completed_orders / nullif(checkout_started, 0), 2) as pct_completed_orders,
        round(100.0 * completed_orders / nullif(sessions, 0), 2) as overall_conversion_rate
    from funnel
)

select * from funnel_metrics

Example 9: Cohort Retention Analysis

-- models/marts/analytics/cohort_retention.sql

with customer_orders as (
    select
        customer_id,
        date_trunc('month', order_date) as order_month
    from {{ ref('fct_orders') }}
    where status = 'completed'
),

first_order as (
    select
        customer_id,
        min(order_month) as cohort_month
    from customer_orders
    group by 1
),

cohort_data as (
    select
        f.cohort_month,
        c.order_month,
        datediff('month', f.cohort_month, c.order_month) as months_since_first_order,
        count(distinct c.customer_id) as customer_count
    from first_order f
    join customer_orders c
        on f.customer_id = c.customer_id
    group by 1, 2, 3
),

cohort_size as (
    select
        cohort_month,
        customer_count as cohort_size
    from cohort_data
    where months_since_first_order = 0
),

retention as (
    select
        cohort_data.cohort_month,
        cohort_data.months_since_first_order,
        cohort_data.customer_count,
        cohort_size.cohort_size,
        round(100.0 * cohort_data.customer_count / cohort_size.cohort_size, 2) as retention_pct
    from cohort_data
    join cohort_size
        on cohort_data.cohort_month = cohort_size.cohort_month
)

select * from retention
order by cohort_month, months_since_first_order

Example 10: Revenue Attribution Model

-- models/marts/analytics/revenue_attribution.sql

with touchpoints as (
    select
        user_id,
        session_id,
        event_timestamp,
        case
            when referrer like '%google%' then 'Google'
            when referrer like '%facebook%' then 'Facebook'
            when referrer like '%email%' then 'Email'
            when referrer is null then 'Direct'
            else 'Other'
        end as channel
    from {{ ref('fct_page_views') }}
),

customer_journeys as (
    select
        t.user_id,
        o.order_id,
        o.order_total,
        t.channel,
        t.event_timestamp,
        o.order_date,
        row_number() over (
            partition by o.order_id
            order by t.event_timestamp
        ) as touchpoint_number,
        count(*) over (partition by o.order_id) as total_touchpoints
    from touchpoints t
    join {{ ref('fct_orders') }} o
        on t.user_id = o.customer_id
        and t.event_timestamp <= o.order_date
        and t.event_timestamp >= dateadd('day', -30, o.order_date)
),

attributed_revenue as (
    select
        order_id,
        channel,
        order_total,
        -- First touch attribution
        case when touchpoint_number = 1
            then order_total else 0 end as first_touch_revenue,
        -- Last touch attribution
        case when touchpoint_number = total_touchpoints
            then order_total else 0 end as last_touch_revenue,
        -- Linear attribution
        order_total / total_touchpoints as linear_revenue,
        -- Time decay (more recent touchpoints get more credit)
        order_total * (power(2, touchpoint_number - 1) /
            (power(2, total_touchpoints) - 1)) as time_decay_revenue
    from customer_journeys
)

select
    channel,
    count(distinct order_id) as orders,
    sum(first_touch_revenue) as first_touch_revenue,
    sum(last_touch_revenue) as last_touch_revenue,
    sum(linear_revenue) as linear_revenue,
    sum(time_decay_revenue) as time_decay_revenue
from attributed_revenue
group by 1

Example 11: Data Quality Test Suite

-- tests/assert_fct_orders_quality.sql

-- Test multiple data quality rules in one test

with order_quality_checks as (
    select
        order_id,
        customer_id,
        order_date,
        order_total,
        status,

        -- Check 1: Order total should be positive
        case when order_total < 0
            then 'Negative order total' end as check_1,

        -- Check 2: Order date should not be in future
        case when order_date > current_date
            then 'Future order date' end as check_2,

        -- Check 3: Customer ID should exist
        case when customer_id is null
            then 'Missing customer ID' end as check_3,

        -- Check 4: Status should be valid
        case when status not in ('placed', 'shipped', 'completed', 'returned', 'cancelled')
            then 'Invalid status' end as check_4

    from {{ ref('fct_orders') }}
),

failed_checks as (
    select
        order_id,
        check_1,
        check_2,
        check_3,
        check_4
    from order_quality_checks
    where check_1 is not null
        or check_2 is not null
        or check_3 is not null
        or check_4 is not null
)

select * from failed_checks

Example 12: Slowly Changing Dimension Merge

-- models/marts/core/dim_products_scd.sql

{{ config(
    materialized='incremental',
    unique_key='product_key',
    merge_update_columns=['product_name', 'category', 'price', 'effective_to', 'is_current']
) }}

with source_data as (
    select
        product_id,
        product_name,
        category,
        price,
        updated_at
    from {{ source('ecommerce', 'products') }}

    {% if is_incremental() %}
        where updated_at > (select max(updated_at) from {{ this }} where is_current = true)
    {% endif %}
),

{% if is_incremental() %}
existing_records as (
    select *
    from {{ this }}
    where is_current = true
),

changed_records as (
    select
        s.product_id,
        s.product_name,
        s.category,
        s.price,
        s.updated_at
    from source_data s
    join existing_records e
        on s.product_id = e.product_id
        and (
            s.product_name != e.product_name
            or s.category != e.category
            or s.price != e.price
        )
),

expire_old_records as (
    select
        e.product_key,
        e.product_id,
        e.product_name,
        e.category,
        e.price,
        e.effective_from,
        c.updated_at as effective_to,
        false as is_current,
        e.updated_at
    from existing_records e
    join changed_records c
        on e.product_id = c.product_id
),

new_versions as (
    select
        {{ dbt_utils.generate_surrogate_key(['c.product_id', 'c.updated_at']) }} as product_key,
        c.product_id,
        c.product_name,
        c.category,
        c.price,
        c.updated_at as effective_from,
        null::timestamp as effective_to,
        true as is_current,
        c.updated_at
    from changed_records c
),

combined as (
    select * from expire_old_records
    union all
    select * from new_versions
)

select * from combined

{% else %}

-- First load: all records are current
select
    {{ dbt_utils.generate_surrogate_key(['product_id', 'updated_at']) }} as product_key,
    product_id,
    product_name,
    category,
    price,
    updated_at as effective_from,
    null::timestamp as effective_to,
    true as is_current,
    updated_at
from source_data

{% endif %}

Example 13: Window Functions for Rankings

-- models/marts/analytics/customer_rfm_score.sql

with customer_metrics as (
    select
        customer_id,
        max(order_date) as last_order_date,
        count(order_id) as total_orders,
        sum(order_total) as total_revenue
    from {{ ref('fct_orders') }}
    where status = 'completed'
    group by 1
),

rfm_calculations as (
    select
        customer_id,
        -- Recency: Days since last order
        datediff('day', last_order_date, current_date) as recency_days,
        -- Frequency: Total orders
        total_orders as frequency,
        -- Monetary: Total revenue
        total_revenue as monetary,

        -- Recency score (1-5, lower days = higher score)
        ntile(5) over (order by datediff('day', last_order_date, current_date) desc) as recency_score,
        -- Frequency score (1-5, more orders = higher score)
        ntile(5) over (order by total_orders) as frequency_score,
        -- Monetary score (1-5, more revenue = higher score)
        ntile(5) over (order by total_revenue) as monetary_score
    from customer_metrics
),

rfm_segments as (
    select
        customer_id,
        recency_days,
        frequency,
        monetary,
        recency_score,
        frequency_score,
        monetary_score,
        recency_score * 100 + frequency_score * 10 + monetary_score as rfm_score,
        case
            when recency_score >= 4 and frequency_score >= 4 and monetary_score >= 4
                then 'Champions'
            when recency_score >= 3 and frequency_score >= 3 and monetary_score >= 3
                then 'Loyal Customers'
            when recency_score >= 4 and frequency_score <= 2 and monetary_score <= 2
                then 'Promising'
            when recency_score >= 3 and frequency_score <= 2 and monetary_score <= 2
                then 'Potential Loyalists'
            when recency_score <= 2 and frequency_score >= 3 and monetary_score >= 3
                then 'At Risk'
            when recency_score <= 2 and frequency_score <= 2 and monetary_score <= 2
                then 'Hibernating'
            when recency_score <= 1
                then 'Lost'
            else 'Need Attention'
        end as customer_segment
    from rfm_calculations
)

select * from rfm_segments

Example 14: Union Multiple Sources

-- models/staging/stg_all_events.sql

{{ config(
    materialized='view'
) }}

-- Union events from multiple sources using dbt_utils

{{
    dbt_utils.union_relations(
        relations=[
            ref('stg_web_events'),
            ref('stg_mobile_events'),
            ref('stg_api_events')
        ],
        exclude=['_loaded_at'],  -- Exclude source-specific columns
        source_column_name='event_source'  -- Add column to track source
    )
}}

Example 15: Surrogate Key Generation

-- models/marts/core/fct_order_lines.sql

with order_lines as (
    select
        order_id,
        line_number,
        product_id,
        quantity,
        unit_price,
        quantity * unit_price as line_total
    from {{ source('ecommerce', 'order_lines') }}
)

select
    {{ dbt_utils.generate_surrogate_key(['order_id', 'line_number']) }} as order_line_key,
    {{ dbt_utils.generate_surrogate_key(['order_id']) }} as order_key,
    {{ dbt_utils.generate_surrogate_key(['product_id']) }} as product_key,
    order_id,
    line_number,
    product_id,
    quantity,
    unit_price,
    line_total
from order_lines

Example 16: Date Spine for Time Series

-- models/marts/analytics/daily_revenue_complete.sql

-- Generate complete date spine to ensure no missing dates

with date_spine as (
    {{ dbt_utils.date_spine(
        datepart="day",
        start_date="cast('2020-01-01' as date)",
        end_date="cast(current_date as date)"
    ) }}
),

daily_revenue as (
    select
        date_trunc('day', order_date) as order_date,
        sum(order_total) as revenue
    from {{ ref('fct_orders') }}
    where status = 'completed'
    group by 1
),

complete_series as (
    select
        date_spine.date_day,
        coalesce(daily_revenue.revenue, 0) as revenue,
        -- 7-day moving average
        avg(coalesce(daily_revenue.revenue, 0)) over (
            order by date_spine.date_day
            rows between 6 preceding and current row
        ) as revenue_7d_ma,
        -- Month-to-date revenue
        sum(coalesce(daily_revenue.revenue, 0)) over (
            partition by date_trunc('month', date_spine.date_day)
            order by date_spine.date_day
        ) as revenue_mtd
    from date_spine
    left join daily_revenue
        on date_spine.date_day = daily_revenue.order_date
)

select * from complete_series

Example 17: Custom Schema Macro Override

-- macros/generate_schema_name.sql

{% macro generate_schema_name(custom_schema_name, node) -%}

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

    {%- if target.name == 'prod' -%}
        -- Production: Use custom schema names directly
        {%- if custom_schema_name is not none -%}
            {{ custom_schema_name | trim }}
        {%- else -%}
            {{ default_schema }}
        {%- endif -%}

    {%- elif target.name == 'dev' -%}
        -- Development: Prefix with dev_username
        {%- if custom_schema_name is not none -%}
            dev_{{ env_var('DBT_USER', 'unknown') }}_{{ custom_schema_name | trim }}
        {%- else -%}
            dev_{{ env_var('DBT_USER', 'unknown') }}
        {%- endif -%}

    {%- else -%}
        -- Default: Concatenate target schema with custom schema
        {{ default_schema }}_{{ custom_schema_name | trim }}
    {%- endif -%}

{%- endmacro %}

Example 18: Cross-Database Query Macro

-- macros/cross_db_concat.sql

-- Handle database-specific concat syntax

{% macro concat(fields) -%}
    {{ return(adapter.dispatch('concat', 'dbt_utils')(fields)) }}
{%- endmacro %}

{% macro default__concat(fields) -%}
    concat({{ fields|join(', ') }})
{%- endmacro %}

{% macro snowflake__concat(fields) -%}
    {{ fields|join(' || ') }}
{%- endmacro %}

{% macro bigquery__concat(fields) -%}
    concat({{ fields|join(', ') }})
{%- endmacro %}

{% macro redshift__concat(fields) -%}
    {{ fields|join(' || ') }}
{%- endmacro %}

Usage:

select
    {{ concat(['first_name', "' '", 'last_name']) }} as full_name
from {{ ref('stg_customers') }}

Example 19: Pre-Hook and Post-Hook Configuration

-- models/marts/core/fct_orders.sql

{{
    config(
        materialized='incremental',
        unique_key='order_id',
        pre_hook=[
            "delete from {{ this }} where order_date < dateadd('year', -3, current_date)",
            "{{ log('Starting incremental load for fct_orders', info=True) }}"
        ],
        post_hook=[
            "create index if not exists idx_fct_orders_customer_id on {{ this }}(customer_id)",
            "create index if not exists idx_fct_orders_order_date on {{ this }}(order_date)",
            "{{ grant_select(this, 'analyst_role') }}",
            "{{ log('Completed incremental load for fct_orders', info=True) }}"
        ],
        tags=['core', 'incremental']
    )
}}

select * from {{ ref('stg_orders') }}

{% if is_incremental() %}
    where order_date > (select max(order_date) from {{ this }})
{% endif %}

Example 20: Exposure Definition

# models/exposures.yml

version: 2

exposures:
  - name: customer_dashboard
    description: |
      Executive dashboard showing customer metrics including:
      - Customer acquisition trends
      - Customer lifetime value
      - Retention rates
      - RFM segmentation
    type: dashboard
    maturity: high
    url: https://looker.company.com/dashboards/customer-metrics

    owner:
      name: Analytics Team
      email: analytics@company.com

    depends_on:
      - ref('fct_orders')
      - ref('dim_customers')
      - ref('customer_rfm_score')
      - ref('cohort_retention')

    tags: ['executive', 'customer-analytics']

  - name: revenue_forecast_model
    description: |
      Machine learning model for revenue forecasting.
      Uses historical order data to predict future revenue.
    type: ml
    maturity: medium
    url: https://mlflow.company.com/models/revenue-forecast

    owner:
      name: Data Science Team
      email: datascience@company.com

    depends_on:
      - ref('fct_orders')
      - ref('daily_revenue_complete')

    tags: ['ml', 'forecasting']

Quick Reference Commands

Essential dbt Commands

# Install dependencies
dbt deps

# Compile project (check for errors)
dbt compile

# Run all models
dbt run

# Run specific model
dbt run --select fct_orders

# Run model and downstream dependencies
dbt run --select fct_orders+

# Run model and upstream dependencies
dbt run --select +fct_orders

# Run model and all dependencies
dbt run --select +fct_orders+

# Run all models in a directory
dbt run --select staging.*

# Run models with specific tag
dbt run --select tag:daily

# Run models, exclude specific ones
dbt run --exclude staging.*

# Run with full refresh (incremental models)
dbt run --full-refresh

# Test all models
dbt test

# Test specific model
dbt test --select fct_orders

# Generate documentation
dbt docs generate

# Serve documentation
dbt docs serve

# Debug connection
dbt debug

# Clean compiled files
dbt clean

# Seed CSV files
dbt seed

# Snapshot models
dbt snapshot

# List resources
dbt ls --select staging.*

# Show compiled SQL
dbt show --select fct_orders

# Parse project
dbt parse

Model Selection Syntax

# By name
--select model_name

# By path
--select staging.jaffle_shop.*

# By tag
--select tag:daily

# By resource type
--select resource_type:model

# By package
--select package:dbt_utils

# By status (modified, new)
--select state:modified+ --state ./prod_manifest

# Combinations (union)
--select model_a model_b

# Intersections
--select tag:daily,staging.*

# Graph operators
--select +model_name    # Upstream dependencies
--select model_name+    # Downstream dependencies
--select +model_name+   # All dependencies
--select @model_name    # Model + children/parents to nth degree

Resources

• Official dbt Documentation: https://docs.getdbt.com/
• dbt Discourse Community: https://discourse.getdbt.com/
• dbt GitHub Repository: https://github.com/dbt-labs/dbt-core
• dbt Package Hub: https://hub.getdbt.com/
• dbt Learn: https://courses.getdbt.com/
• dbt Style Guide: https://github.com/dbt-labs/corp/blob/main/dbt_style_guide.md
• Analytics Engineering Guide: https://www.getdbt.com/analytics-engineering/
• dbt Slack Community: https://www.getdbt.com/community/join-the-community/

---

Skill Version: 1.0.0 Last Updated: October 2025 Skill Category: Data Engineering, Analytics Engineering, Data Transformation Compatible With: dbt Core 1.0+, dbt Cloud, Snowflake, BigQuery, Redshift, Postgres, Databricks

dbt Data Transformation Examples

Practical, production-ready examples for building data transformation pipelines with dbt. Each example includes complete code, explanations, and best practices.

Table of Contents

1. Staging Models 2. Fact Tables 3. Dimension Tables 4. Incremental Models 5. Testing Patterns 6. Documentation 7. Macros 8. Snapshots 9. Advanced Analytics 10. Production Workflows

---

1. Staging Models

Example 1.1: Basic Staging Model

Clean and rename raw data from sources:

-- models/staging/jaffle_shop/stg_jaffle_shop__customers.sql

{{
    config(
        materialized='view',
        tags=['staging', 'daily']
    )
}}

with source as (
    -- Use source() to reference raw tables
    select * from {{ source('jaffle_shop', 'customers') }}
),

renamed as (
    select
        -- Rename for clarity and consistency
        id as customer_id,
        first_name,
        last_name,

        -- Create computed columns
        first_name || ' ' || last_name as customer_name,
        lower(trim(email)) as email,

        -- Preserve audit columns
        created_at,
        updated_at,
        _loaded_at

    from source
)

select * from renamed

Configuration:

# models/staging/jaffle_shop/_jaffle_shop__sources.yml

version: 2

sources:
  - name: jaffle_shop
    description: Raw data from Jaffle Shop application database
    database: raw
    schema: jaffle_shop

    tables:
      - name: customers
        description: Customer records from the application
        loaded_at_field: _loaded_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}

        columns:
          - name: id
            description: Primary key
            tests:
              - unique
              - not_null
          - name: email
            description: Customer email address
            tests:
              - not_null

Example 1.2: Staging with Type Casting

Handle data type conversions and parsing:

-- models/staging/stripe/stg_stripe__payments.sql

with source as (
    select * from {{ source('stripe', 'payments') }}
),

cleaned as (
    select
        id as payment_id,
        order_id,
        payment_method,

        -- Type casting
        cast(amount as decimal(10,2)) as amount,
        cast(created_at as timestamp) as payment_timestamp,

        -- Parse JSON columns
        parse_json(metadata):customer_ip::string as customer_ip,
        parse_json(metadata):user_agent::string as user_agent,

        -- Standardize status values
        case lower(trim(status))
            when 'success' then 'succeeded'
            when 'fail' then 'failed'
            else lower(trim(status))
        end as payment_status,

        -- Handle null/empty strings
        nullif(trim(failure_message), '') as failure_message,

        _loaded_at

    from source
),

validated as (
    select *
    from cleaned

    -- Filter out invalid records
    where payment_id is not null
        and order_id is not null
        and amount >= 0
)

select * from validated

Example 1.3: Staging with Deduplication

Handle duplicate records in source data:

-- models/staging/events/stg_events__page_views.sql

{{
    config(
        materialized='view'
    )
}}

with source as (
    select * from {{ source('events', 'page_views') }}
),

deduplicated as (
    select
        event_id,
        user_id,
        session_id,
        event_timestamp,
        page_path,
        referrer,

        -- Use row_number to identify duplicates
        row_number() over (
            partition by event_id
            order by _loaded_at desc
        ) as row_num

    from source
),

final as (
    select
        event_id,
        user_id,
        session_id,
        event_timestamp,
        page_path,
        referrer
    from deduplicated
    where row_num = 1  -- Keep most recent version
)

select * from final

---

2. Fact Tables

Example 2.1: Order Fact Table

Build a comprehensive fact table with measures and foreign keys:

-- models/marts/core/fct_orders.sql

{{
    config(
        materialized='table',
        partition_by={
            'field': 'order_date',
            'data_type': 'date',
            'granularity': 'day'
        },
        cluster_by=['customer_id', 'status'],
        tags=['core', 'daily']
    )
}}

with orders as (
    select * from {{ ref('stg_jaffle_shop__orders') }}
),

customers as (
    select * from {{ ref('dim_customers') }}
),

payments as (
    select
        order_id,
        sum(case when payment_status = 'succeeded' then amount else 0 end) as total_paid,
        sum(amount) as total_attempted,
        count(*) as payment_count,
        max(payment_timestamp) as last_payment_timestamp
    from {{ ref('stg_stripe__payments') }}
    group by 1
),

order_items as (
    select
        order_id,
        count(*) as item_count,
        sum(quantity) as total_quantity
    from {{ ref('stg_jaffle_shop__order_items') }}
    group by 1
),

final as (
    select
        -- Primary key
        orders.order_id,

        -- Foreign keys
        orders.customer_id,
        customers.customer_segment,

        -- Dates
        orders.order_date,
        date_trunc('month', orders.order_date) as order_month,
        date_trunc('year', orders.order_date) as order_year,

        -- Order attributes
        orders.status,

        -- Measures (additive facts)
        coalesce(payments.total_paid, 0) as order_total,
        coalesce(payments.total_attempted, 0) as amount_attempted,
        coalesce(order_items.item_count, 0) as line_item_count,
        coalesce(order_items.total_quantity, 0) as total_quantity,

        -- Semi-additive facts
        case when payments.payment_count > 1 then 1 else 0 end as has_multiple_payments,

        -- Flags and indicators
        case when orders.status = 'completed' then 1 else 0 end as is_completed,
        case when orders.status = 'returned' then 1 else 0 end as is_returned,

        -- Timestamps
        orders.created_at as order_created_at,
        payments.last_payment_timestamp,

        -- Audit columns
        current_timestamp() as dbt_updated_at

    from orders
    left join customers
        on orders.customer_id = customers.customer_id
    left join payments
        on orders.order_id = payments.order_id
    left join order_items
        on orders.order_id = order_items.order_id
)

select * from final

Example 2.2: Event Fact Table (Many-to-Many)

Handle events with multiple dimensions:

-- models/marts/analytics/fct_user_events.sql

{{
    config(
        materialized='incremental',
        unique_key='event_id',
        partition_by={
            'field': 'event_date',
            'data_type': 'date'
        },
        cluster_by=['user_id', 'event_type']
    )
}}

with events as (
    select * from {{ ref('stg_events__raw_events') }}

    {% if is_incremental() %}
        where event_timestamp > (select max(event_timestamp) from {{ this }})
    {% endif %}
),

users as (
    select * from {{ ref('dim_users') }}
),

sessions as (
    select * from {{ ref('dim_sessions') }}
),

final as (
    select
        -- Primary key
        events.event_id,

        -- Foreign keys (multiple dimensions)
        events.user_id,
        events.session_id,
        events.device_id,
        events.page_id,

        -- Degenerate dimensions (stored in fact)
        events.event_type,
        events.event_category,

        -- Date/time dimensions
        events.event_timestamp,
        date(events.event_timestamp) as event_date,
        date_trunc('hour', events.event_timestamp) as event_hour,

        -- Measures
        coalesce(events.event_value, 0) as event_value,
        events.duration_seconds,

        -- User attributes (from dimension)
        users.user_segment,
        users.acquisition_channel,

        -- Session attributes
        sessions.is_first_session,
        sessions.device_type,

        -- Flags
        case when events.event_type = 'purchase' then 1 else 0 end as is_conversion_event,

        -- Audit
        current_timestamp() as dbt_updated_at

    from events
    left join users
        on events.user_id = users.user_id
    left join sessions
        on events.session_id = sessions.session_id
)

select * from final

---

3. Dimension Tables

Example 3.1: Customer Dimension with Enrichment

Create a slowly changing dimension with derived attributes:

-- models/marts/core/dim_customers.sql

{{
    config(
        materialized='table',
        tags=['core', 'dimension']
    )
}}

with customers as (
    select * from {{ ref('stg_jaffle_shop__customers') }}
),

customer_orders as (
    select
        customer_id,
        min(order_date) as first_order_date,
        max(order_date) as most_recent_order_date,
        count(distinct order_id) as total_orders,
        count(distinct case when status = 'completed' then order_id end) as completed_orders,
        sum(case when status = 'completed' then order_total else 0 end) as lifetime_value,
        avg(case when status = 'completed' then order_total end) as avg_order_value
    from {{ ref('fct_orders') }}
    group by 1
),

customer_segments as (
    select
        customer_id,
        case
            when lifetime_value >= 1000 then 'VIP'
            when lifetime_value >= 500 then 'High Value'
            when total_orders >= 5 then 'Regular'
            when total_orders >= 1 then 'New'
            else 'Prospect'
        end as customer_segment,

        case
            when most_recent_order_date >= current_date - interval '30 days' then 'Active'
            when most_recent_order_date >= current_date - interval '90 days' then 'At Risk'
            when most_recent_order_date < current_date - interval '90 days' then 'Churned'
            else 'Prospect'
        end as customer_status
    from customer_orders
),

final as (
    select
        -- Surrogate key (optional, for SCD Type 2)
        {{ dbt_utils.generate_surrogate_key(['customers.customer_id']) }} as customer_key,

        -- Natural key
        customers.customer_id,

        -- Attributes
        customers.customer_name,
        customers.first_name,
        customers.last_name,
        customers.email,

        -- Derived attributes
        coalesce(customer_orders.first_order_date, null) as first_order_date,
        coalesce(customer_orders.most_recent_order_date, null) as most_recent_order_date,
        coalesce(customer_orders.total_orders, 0) as total_orders,
        coalesce(customer_orders.completed_orders, 0) as completed_orders,
        coalesce(customer_orders.lifetime_value, 0) as lifetime_value,
        coalesce(customer_orders.avg_order_value, 0) as avg_order_value,

        -- Calculated metrics
        datediff('day', customer_orders.first_order_date, customer_orders.most_recent_order_date) as customer_tenure_days,
        case
            when customer_orders.total_orders > 0
            then customer_orders.lifetime_value / customer_orders.total_orders
            else 0
        end as avg_order_size,

        -- Segments
        coalesce(customer_segments.customer_segment, 'Prospect') as customer_segment,
        coalesce(customer_segments.customer_status, 'Prospect') as customer_status,

        -- Flags
        case when customer_orders.total_orders > 0 then true else false end as has_ordered,
        case when customer_orders.completed_orders > 0 then true else false end as has_completed_order,

        -- Timestamps
        customers.created_at as customer_created_at,
        customers.updated_at as customer_updated_at,

        -- Audit
        current_timestamp() as dbt_updated_at

    from customers
    left join customer_orders
        on customers.customer_id = customer_orders.customer_id
    left join customer_segments
        on customers.customer_id = customer_segments.customer_id
)

select * from final

Example 3.2: Date Dimension

Generate a comprehensive date dimension table:

-- models/marts/core/dim_date.sql

{{
    config(
        materialized='table',
        tags=['dimension', 'reference']
    )
}}

with date_spine as (
    -- Generate dates for the next 10 years
    {{ dbt_utils.date_spine(
        datepart="day",
        start_date="cast('2020-01-01' as date)",
        end_date="cast(dateadd('year', 10, current_date) as date)"
    ) }}
),

date_attributes as (
    select
        date_day,

        -- Date parts
        extract(year from date_day) as year_number,
        extract(quarter from date_day) as quarter_number,
        extract(month from date_day) as month_number,
        extract(week from date_day) as week_number,
        extract(dayofyear from date_day) as day_of_year,
        extract(dayofweek from date_day) as day_of_week,
        extract(dayofmonth from date_day) as day_of_month,

        -- Date names
        to_char(date_day, 'YYYY') as year_name,
        to_char(date_day, 'YYYY-Q') as quarter_name,
        to_char(date_day, 'YYYY-MM') as month_name,
        to_char(date_day, 'Mon') as month_short_name,
        to_char(date_day, 'Month') as month_long_name,
        to_char(date_day, 'Dy') as day_short_name,
        to_char(date_day, 'Day') as day_long_name,

        -- Fiscal periods (assuming fiscal year starts July 1)
        case
            when extract(month from date_day) >= 7
            then extract(year from date_day) + 1
            else extract(year from date_day)
        end as fiscal_year,
        case
            when extract(month from date_day) between 7 and 9 then 1
            when extract(month from date_day) between 10 and 12 then 2
            when extract(month from date_day) between 1 and 3 then 3
            when extract(month from date_day) between 4 and 6 then 4
        end as fiscal_quarter,

        -- Flags
        case when extract(dayofweek from date_day) in (0, 6) then true else false end as is_weekend,
        case when extract(dayofweek from date_day) between 1 and 5 then true else false end as is_weekday,

        -- Relative dates
        case when date_day = current_date then true else false end as is_today,
        case when date_day = current_date - interval '1 day' then true else false end as is_yesterday,
        case when date_trunc('week', date_day) = date_trunc('week', current_date) then true else false end as is_current_week,
        case when date_trunc('month', date_day) = date_trunc('month', current_date) then true else false end as is_current_month,
        case when date_trunc('quarter', date_day) = date_trunc('quarter', current_date) then true else false end as is_current_quarter,
        case when date_trunc('year', date_day) = date_trunc('year', current_date) then true else false end as is_current_year,

        -- Period start/end
        date_trunc('week', date_day) as week_start_date,
        date_trunc('month', date_day) as month_start_date,
        date_trunc('quarter', date_day) as quarter_start_date,
        date_trunc('year', date_day) as year_start_date

    from date_spine
)

select * from date_attributes

---

4. Incremental Models

Example 4.1: Append-Only Incremental

For immutable event data:

-- models/marts/analytics/fct_page_views_incremental.sql

{{
    config(
        materialized='incremental',
        unique_key='page_view_id',
        incremental_strategy='append',
        partition_by={
            'field': 'event_date',
            'data_type': 'date',
            'granularity': 'day'
        },
        cluster_by=['user_id', 'session_id']
    )
}}

with page_views as (
    select
        event_id as page_view_id,
        user_id,
        session_id,
        event_timestamp,
        date(event_timestamp) as event_date,
        page_path,
        referrer,
        device_type,
        _loaded_at

    from {{ ref('stg_events__page_views') }}

    {% if is_incremental() %}
        -- Use _loaded_at to catch late-arriving data
        where _loaded_at > (
            select coalesce(max(_loaded_at), '1900-01-01'::timestamp)
            from {{ this }}
        )
    {% endif %}
)

select * from page_views

Example 4.2: Merge Incremental with Updates

For data that can change:

-- models/marts/core/fct_orders_incremental.sql

{{
    config(
        materialized='incremental',
        unique_key='order_id',
        incremental_strategy='merge',
        merge_update_columns=['status', 'order_total', 'updated_at'],
        on_schema_change='fail'
    )
}}

with orders as (
    select
        order_id,
        customer_id,
        order_date,
        status,
        order_total,
        created_at,
        updated_at

    from {{ ref('stg_jaffle_shop__orders') }}

    {% if is_incremental() %}
        -- Look back 3 days to catch status updates
        where updated_at > (
            select dateadd('day', -3, max(updated_at))
            from {{ this }}
        )
    {% endif %}
)

select * from orders

Example 4.3: Delete+Insert Incremental

For daily aggregations:

-- models/marts/analytics/daily_metrics.sql

{{
    config(
        materialized='incremental',
        unique_key=['metric_date', 'customer_segment'],
        incremental_strategy='delete+insert',
        partition_by={
            'field': 'metric_date',
            'data_type': 'date'
        }
    )
}}

with orders as (
    select
        date_trunc('day', order_date) as order_date,
        customer_id,
        status,
        order_total
    from {{ ref('fct_orders') }}

    {% if is_incremental() %}
        -- Reprocess last 7 days to handle late updates
        where date_trunc('day', order_date) >= (
            select max(metric_date) - interval '7 days'
            from {{ this }}
        )
    {% endif %}
),

customers as (
    select
        customer_id,
        customer_segment
    from {{ ref('dim_customers') }}
),

daily_metrics as (
    select
        orders.order_date as metric_date,
        customers.customer_segment,
        count(distinct orders.order_id) as order_count,
        count(distinct orders.customer_id) as unique_customers,
        sum(case when orders.status = 'completed' then orders.order_total else 0 end) as revenue,
        avg(case when orders.status = 'completed' then orders.order_total end) as avg_order_value

    from orders
    left join customers
        on orders.customer_id = customers.customer_id
    group by 1, 2
)

select * from daily_metrics

---

5. Testing Patterns

Example 5.1: Comprehensive Schema Tests

# models/marts/core/_core__models.yml

version: 2

models:
  - name: fct_orders
    description: Order fact table

    tests:
      # Table-level tests
      - dbt_utils.expression_is_true:
          expression: "order_total >= 0"
          config:
            severity: error

      - dbt_utils.unique_combination_of_columns:
          combination_of_columns:
            - order_id
            - order_date
          config:
            severity: warn

    columns:
      - name: order_id
        description: Primary key
        tests:
          - unique:
              config:
                severity: error
          - not_null:
              config:
                severity: error

      - name: customer_id
        description: Foreign key to dim_customers
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
              config:
                severity: error

      - name: status
        description: Order status
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned', 'cancelled']
              config:
                severity: error

      - name: order_total
        description: Total order amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

      - name: order_date
        description: Date order was placed
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: "<= current_date"
              config:
                error_if: ">100"
                warn_if: ">0"

Example 5.2: Custom Data Quality Tests

-- tests/assert_revenue_reconciliation.sql

-- Ensure revenue in fct_orders matches payments

with order_revenue as (
    select
        sum(order_total) as total_from_orders
    from {{ ref('fct_orders') }}
    where status = 'completed'
),

payment_revenue as (
    select
        sum(amount) as total_from_payments
    from {{ ref('stg_stripe__payments') }}
    where payment_status = 'succeeded'
),

reconciliation as (
    select
        order_revenue.total_from_orders,
        payment_revenue.total_from_payments,
        abs(order_revenue.total_from_orders - payment_revenue.total_from_payments) as difference,
        order_revenue.total_from_orders * 0.01 as threshold  -- 1% tolerance
    from order_revenue
    cross join payment_revenue
)

-- Test fails if difference exceeds threshold
select *
from reconciliation
where difference > threshold

-- tests/assert_no_future_dates.sql

-- Ensure no records have dates in the future

select
    'fct_orders' as table_name,
    order_id as record_id,
    order_date as problematic_date
from {{ ref('fct_orders') }}
where order_date > current_date

union all

select
    'fct_page_views' as table_name,
    page_view_id as record_id,
    event_timestamp::date as problematic_date
from {{ ref('fct_page_views') }}
where event_timestamp > current_timestamp

Example 5.3: Cross-Model Consistency Tests

-- tests/assert_customer_order_consistency.sql

-- Ensure customer metrics in dim_customers match fct_orders

with customer_orders_from_dim as (
    select
        customer_id,
        total_orders as order_count_from_dim,
        lifetime_value as ltv_from_dim
    from {{ ref('dim_customers') }}
),

customer_orders_from_fact as (
    select
        customer_id,
        count(distinct order_id) as order_count_from_fact,
        sum(case when status = 'completed' then order_total else 0 end) as ltv_from_fact
    from {{ ref('fct_orders') }}
    group by 1
),

comparison as (
    select
        dim.customer_id,
        dim.order_count_from_dim,
        fact.order_count_from_fact,
        dim.ltv_from_dim,
        fact.ltv_from_fact
    from customer_orders_from_dim dim
    full outer join customer_orders_from_fact fact
        on dim.customer_id = fact.customer_id
)

-- Fail if counts or values don't match
select *
from comparison
where order_count_from_dim != order_count_from_fact
    or abs(ltv_from_dim - ltv_from_fact) > 0.01

---

6. Documentation

Example 6.1: Comprehensive Model Documentation

# models/marts/core/_core__models.yml

version: 2

models:
  - name: fct_orders
    description: |
      # Order Fact Table

      This table contains one row per order with associated customer,
      payment, and product information.

      ## Grain
      One row per order (order_id is unique)

      ## Refresh Schedule
      - **Development**: On-demand via dbt Cloud IDE
      - **Production**: Daily at 2:00 AM UTC
      - **Incremental**: Processes last 3 days of data

      ## Business Logic
      - Only includes orders from the Jaffle Shop application
      - Order totals calculated from successful Stripe payments
      - Status reflects current order state (may change over time)

      ## Data Quality
      - All orders must have a valid customer_id
      - Order totals must be non-negative
      - Order dates cannot be in the future

      ## Usage Examples

-- Get total revenue by month select date_trunc('month', order_date) as month, sum(order_total) as revenue from {{ ref('fct_orders') }} where status = 'completed' group by 1;

      ## Known Issues
      - Guest checkouts may have temporary customer_id values
      - Cancelled orders within 24 hours may show as 'completed' briefly

      ## Related Models
      - Upstream: {{ ref('stg_jaffle_shop__orders') }}, {{ ref('stg_stripe__payments') }}
      - Downstream: {{ ref('daily_revenue_metrics') }}, {{ ref('customer_lifetime_value') }}

    meta:
      owner: analytics_team@company.com
      contains_pii: true
      pii_columns: [customer_id]

    columns:
      - name: order_id
        description: |
          **Primary key** for the orders table.

          Uniquely identifies each order. Generated by the application
          at order creation time. Format: alphanumeric, 32 characters.

        tests:
          - unique
          - not_null

      - name: customer_id
        description: |
          **Foreign key** to {{ ref('dim_customers') }}.

          Links each order to the customer who placed it.

          **Note**: May be NULL for guest checkout orders (rare).

        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id

      - name: order_date
        description: Date the order was placed (UTC timezone)
        tests:
          - not_null

      - name: status
        description: "{{ doc('order_status') }}"
        tests:
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned', 'cancelled']

      - name: order_total
        description: |
          Total order amount in USD, including:
          - Product costs
          - Shipping fees
          - Taxes
          - Discounts (subtracted)

          Calculated from successful Stripe payment records.

        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

Example 6.2: Documentation Blocks

<!-- models/docs.md -->

{% docs order_status %}

### Order Status Values

The current fulfillment status of an order.

| Status | Description | Transition Conditions |
|--------|-------------|----------------------|
| `placed` | Order received, payment pending | Initial state |
| `shipped` | Order dispatched to customer | After payment confirmed |
| `completed` | Order delivered successfully | After delivery confirmation |
| `returned` | Customer returned the order | Within 30-day window |
| `cancelled` | Order cancelled before shipment | Customer or system initiated |

**Lifecycle Flow:**

placed → shipped → completed ↓ ↓ cancelled returned

**Business Rules:**
- Orders can only be cancelled in 'placed' status
- Returns accepted within 30 days of delivery
- Status changes tracked in order_history table

{% enddocs %}

{% docs customer_segment %}

### Customer Segmentation Logic

Customers are categorized into segments based on lifetime value and order history:

- **VIP**: Lifetime value ≥ $1,000
- **High Value**: Lifetime value ≥ $500 and < $1,000
- **Regular**: 5+ orders regardless of value
- **New**: 1-4 orders
- **Prospect**: No completed orders

Segments recalculated daily as part of {{ ref('dim_customers') }} refresh.

{% enddocs %}

{% docs data_freshness %}

### Data Freshness Expectations

| Source | Refresh Frequency | Acceptable Lag |
|--------|------------------|----------------|
| Application DB | Real-time CDC | < 5 minutes |
| Stripe API | Hourly sync | < 2 hours |
| Google Analytics | Daily batch | < 24 hours |

Source freshness monitored via dbt source freshness tests.

{% enddocs %}

---

7. Macros

Example 7.1: Reusable Calculation Macro

-- macros/cents_to_dollars.sql

{% macro cents_to_dollars(column_name, precision=2) %}
    round(cast({{ column_name }} as numeric) / 100.0, {{ precision }})
{% endmacro %}

Usage:

select
    payment_id,
    {{ cents_to_dollars('amount_cents') }} as amount_dollars,
    {{ cents_to_dollars('amount_cents', 4) }} as amount_dollars_precise
from {{ ref('stg_stripe__payments') }}

Example 7.2: Dynamic Pivot Macro

-- macros/pivot_metric.sql

{% macro pivot_metric(table, group_by_col, pivot_col, metric_col, agg='sum', prefix='', suffix='') %}

{% set pivot_values_query %}
    select distinct {{ pivot_col }}
    from {{ table }}
    where {{ pivot_col }} is not null
    order by {{ pivot_col }}
{% endset %}

{% set results = run_query(pivot_values_query) %}

{% if execute %}
    {% set pivot_values = results.columns[0].values() %}
{% else %}
    {% set pivot_values = [] %}
{% endif %}

select
    {{ group_by_col }},
    {% for value in pivot_values %}
        {{ agg }}(case when {{ pivot_col }} = '{{ value }}' then {{ metric_col }} else 0 end)
            as {{ prefix }}{{ value | replace(' ', '_') | lower }}{{ suffix }}
        {% if not loop.last %},{% endif %}
    {% endfor %}
from {{ table }}
group by {{ group_by_col }}

{% endmacro %}

Usage:

-- Pivot revenue by order status
{{ pivot_metric(
    table=ref('fct_orders'),
    group_by_col='customer_id',
    pivot_col='status',
    metric_col='order_total',
    agg='sum',
    suffix='_revenue'
) }}

Example 7.3: Grant Permissions Macro

-- macros/grant_permissions.sql

{% macro grant_select(schema, role) %}

    {% if target.name == 'prod' %}
        {% set sql %}
            grant select on all tables in schema {{ schema }} to role {{ role }};
            grant select on all views in schema {{ schema }} to role {{ role }};
            grant select on future tables in schema {{ schema }} to role {{ role }};
            grant select on future views in schema {{ schema }} to role {{ role }};
        {% endset %}

        {% do run_query(sql) %}
        {% do log("Granted SELECT on " ~ schema ~ " to " ~ role, info=True) %}
    {% else %}
        {% do log("Skipping grants in " ~ target.name ~ " environment", info=True) %}
    {% endif %}

{% endmacro %}

Usage in dbt_project.yml:

on-run-end:
  - "{{ grant_select(target.schema, 'analyst_role') }}"

Example 7.4: Audit Columns Macro

-- macros/audit_columns.sql

{% macro audit_columns() %}
    current_timestamp() as dbt_updated_at,
    '{{ invocation_id }}' as dbt_invocation_id,
    '{{ var("dbt_user", "system") }}' as dbt_updated_by
{% endmacro %}

Usage:

select
    order_id,
    customer_id,
    order_total,
    {{ audit_columns() }}
from {{ ref('stg_orders') }}

---

8. Snapshots

Example 8.1: Timestamp Strategy Snapshot

-- snapshots/customers_snapshot.sql

{% snapshot customers_snapshot %}

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

select
    customer_id,
    customer_name,
    email,
    customer_segment,
    customer_status,
    lifetime_value,
    updated_at
from {{ ref('dim_customers') }}

{% endsnapshot %}

Result includes dbt-generated columns:

• dbt_valid_from: When record became active
• dbt_valid_to: When record was superseded (NULL if current)
• dbt_updated_at: Snapshot run timestamp

Example 8.2: Check Strategy Snapshot

-- snapshots/product_prices_snapshot.sql

{% snapshot product_prices_snapshot %}

{{
    config(
        target_schema='snapshots',
        unique_key='product_id',
        strategy='check',
        check_cols=['price', 'discount_pct', 'is_active']
    )
}}

select
    product_id,
    product_name,
    category,
    price,
    discount_pct,
    is_active
from {{ source('ecommerce', 'products') }}

{% endsnapshot %}

---

9. Advanced Analytics

Example 9.1: Cohort Retention Analysis

-- models/marts/analytics/customer_retention_cohorts.sql

with customer_first_order as (
    select
        customer_id,
        date_trunc('month', min(order_date)) as cohort_month
    from {{ ref('fct_orders') }}
    where status = 'completed'
    group by 1
),

customer_orders_by_month as (
    select
        customer_id,
        date_trunc('month', order_date) as order_month
    from {{ ref('fct_orders') }}
    where status = 'completed'
    group by 1, 2
),

cohort_activity as (
    select
        f.cohort_month,
        o.order_month,
        datediff('month', f.cohort_month, o.order_month) as months_since_first_order,
        count(distinct o.customer_id) as active_customers
    from customer_first_order f
    join customer_orders_by_month o
        on f.customer_id = o.customer_id
    group by 1, 2, 3
),

cohort_size as (
    select
        cohort_month,
        count(distinct customer_id) as cohort_size
    from customer_first_order
    group by 1
),

retention_rates as (
    select
        a.cohort_month,
        a.months_since_first_order,
        s.cohort_size,
        a.active_customers,
        round(100.0 * a.active_customers / s.cohort_size, 2) as retention_pct
    from cohort_activity a
    join cohort_size s
        on a.cohort_month = s.cohort_month
)

select * from retention_rates
order by cohort_month, months_since_first_order

Example 9.2: RFM Segmentation

-- models/marts/analytics/customer_rfm_analysis.sql

with customer_metrics as (
    select
        customer_id,
        max(order_date) as last_order_date,
        count(distinct order_id) as frequency,
        sum(order_total) as monetary
    from {{ ref('fct_orders') }}
    where status = 'completed'
    group by 1
),

rfm_scores as (
    select
        customer_id,
        datediff('day', last_order_date, current_date) as recency_days,
        frequency,
        monetary,

        -- Score 1-5 (5 = best)
        ntile(5) over (order by last_order_date desc) as recency_score,
        ntile(5) over (order by frequency) as frequency_score,
        ntile(5) over (order by monetary) as monetary_score
    from customer_metrics
),

rfm_segments as (
    select
        *,
        recency_score * 100 + frequency_score * 10 + monetary_score as rfm_combined_score,
        case
            -- Champions: Bought recently, buy often, spend the most
            when recency_score >= 4 and frequency_score >= 4 and monetary_score >= 4
                then 'Champions'

            -- Loyal Customers: Buy regularly, good spenders
            when recency_score >= 3 and frequency_score >= 3 and monetary_score >= 3
                then 'Loyal Customers'

            -- Potential Loyalists: Recent customers, spent good amount, bought more than once
            when recency_score >= 4 and frequency_score >= 2 and monetary_score >= 2
                then 'Potential Loyalists'

            -- Recent Customers: Bought recently, but not often
            when recency_score >= 4 and frequency_score <= 2
                then 'Recent Customers'

            -- Promising: Recent shoppers, but haven't spent much
            when recency_score >= 3 and frequency_score <= 2 and monetary_score <= 2
                then 'Promising'

            -- Need Attention: Above average recency, frequency, and monetary values
            when recency_score >= 3 and frequency_score >= 2 and monetary_score >= 2
                then 'Need Attention'

            -- About to Sleep: Below average recency, frequency, and monetary values
            when recency_score <= 2 and frequency_score >= 2 and monetary_score >= 2
                then 'About To Sleep'

            -- At Risk: Spent big money, purchased often, but long time ago
            when recency_score <= 2 and frequency_score >= 3 and monetary_score >= 3
                then 'At Risk'

            -- Cannot Lose Them: Made big purchases, often, but haven't returned for long time
            when recency_score <= 1 and frequency_score >= 4 and monetary_score >= 4
                then 'Cannot Lose Them'

            -- Hibernating: Last purchase long ago, low spenders, low frequency
            when recency_score <= 2 and frequency_score <= 2 and monetary_score <= 2
                then 'Hibernating'

            -- Lost: Lowest recency, frequency, and monetary scores
            when recency_score <= 1
                then 'Lost'

            else 'Other'
        end as rfm_segment

    from rfm_scores
)

select * from rfm_segments

Example 9.3: Marketing Attribution

-- models/marts/analytics/marketing_attribution.sql

with touchpoints as (
    select
        user_id,
        session_id,
        event_timestamp,
        case
            when referrer like '%google%' then 'Google'
            when referrer like '%facebook%' then 'Facebook'
            when referrer like '%instagram%' then 'Instagram'
            when referrer like '%email%' then 'Email'
            when referrer is null then 'Direct'
            else 'Other'
        end as channel
    from {{ ref('fct_page_views') }}
    where user_id is not null
),

conversions as (
    select
        customer_id as user_id,
        order_id,
        order_date,
        order_total
    from {{ ref('fct_orders') }}
    where status = 'completed'
),

customer_journey as (
    select
        c.order_id,
        c.order_total,
        t.channel,
        t.event_timestamp as touchpoint_timestamp,
        c.order_date,

        -- Touchpoint sequencing
        row_number() over (
            partition by c.order_id
            order by t.event_timestamp
        ) as touchpoint_number,

        count(*) over (partition by c.order_id) as total_touchpoints,

        -- Time decay weight (more recent = higher weight)
        datediff('hour', t.event_timestamp, c.order_date) as hours_before_conversion

    from conversions c
    join touchpoints t
        on c.user_id = t.user_id
        and t.event_timestamp <= c.order_date
        and t.event_timestamp >= dateadd('day', -30, c.order_date)
),

attributed_revenue as (
    select
        order_id,
        channel,
        order_total,
        touchpoint_number,
        total_touchpoints,
        hours_before_conversion,

        -- First Touch Attribution
        case when touchpoint_number = 1
            then order_total else 0 end as first_touch_revenue,

        -- Last Touch Attribution
        case when touchpoint_number = total_touchpoints
            then order_total else 0 end as last_touch_revenue,

        -- Linear Attribution (equal credit to all touchpoints)
        order_total / total_touchpoints as linear_revenue,

        -- Time Decay Attribution (exponential decay, 7-day half-life)
        order_total * exp(-0.1 * (hours_before_conversion / 24.0))
            / sum(exp(-0.1 * (hours_before_conversion / 24.0))) over (partition by order_id)
            as time_decay_revenue,

        -- Position-Based Attribution (40% first, 40% last, 20% middle)
        case
            when total_touchpoints = 1 then order_total
            when touchpoint_number = 1 then order_total * 0.4
            when touchpoint_number = total_touchpoints then order_total * 0.4
            else order_total * 0.2 / (total_touchpoints - 2)
        end as position_based_revenue

    from customer_journey
)

select
    channel,
    count(distinct order_id) as orders,
    sum(first_touch_revenue) as first_touch_revenue,
    sum(last_touch_revenue) as last_touch_revenue,
    sum(linear_revenue) as linear_revenue,
    sum(time_decay_revenue) as time_decay_revenue,
    sum(position_based_revenue) as position_based_revenue
from attributed_revenue
group by 1

---

10. Production Workflows

Example 10.1: CI/CD with GitHub Actions

# .github/workflows/dbt_ci.yml

name: dbt CI

on:
  pull_request:
    branches: [main]
    paths:
      - 'models/**'
      - 'macros/**'
      - 'tests/**'
      - 'dbt_project.yml'
      - 'packages.yml'

jobs:
  dbt_ci_check:
    runs-on: ubuntu-latest

    env:
      DBT_PROFILES_DIR: .
      DBT_SNOWFLAKE_ACCOUNT: ${{ secrets.DBT_SNOWFLAKE_ACCOUNT }}
      DBT_SNOWFLAKE_USER: ${{ secrets.DBT_CI_USER }}
      DBT_SNOWFLAKE_PASSWORD: ${{ secrets.DBT_CI_PASSWORD }}
      DBT_SNOWFLAKE_ROLE: TRANSFORMER
      DBT_SNOWFLAKE_DATABASE: ANALYTICS
      DBT_SNOWFLAKE_WAREHOUSE: TRANSFORMING
      DBT_SNOWFLAKE_SCHEMA: dbt_ci_${{ github.event.pull_request.number }}

    steps:
      - name: Checkout code
        uses: actions/checkout@v3
        with:
          fetch-depth: 0  # Needed for state comparison

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dbt
        run: |
          pip install dbt-core==1.7.4 dbt-snowflake==1.7.1

      - name: Install dbt packages
        run: dbt deps

      - name: Download production manifest
        run: |
          mkdir -p ./prod_state
          # Download manifest.json from S3 or artifact storage
          # aws s3 cp s3://your-bucket/prod/manifest.json ./prod_state/

      - name: dbt debug
        run: dbt debug --target ci

      - name: dbt compile (all models)
        run: dbt compile --target ci

      - name: dbt run (modified models only)
        run: |
          dbt run \
            --select state:modified+ \
            --state ./prod_state \
            --target ci

      - name: dbt test (modified models only)
        run: |
          dbt test \
            --select state:modified+ \
            --state ./prod_state \
            --target ci

      - name: Generate dbt docs
        run: dbt docs generate --target ci

      - name: Comment PR with results
        uses: actions/github-script@v6
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: '✅ dbt CI checks passed! Models compiled and tests succeeded.'
            })

      - name: Cleanup CI schema
        if: always()
        run: |
          dbt run-operation drop_schema \
            --args '{schema: dbt_ci_${{ github.event.pull_request.number }}}' \
            --target ci

Example 10.2: Production Deployment

# .github/workflows/dbt_production.yml

name: dbt Production Deploy

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 2 * * *'  # Daily at 2 AM UTC

jobs:
  deploy_production:
    runs-on: ubuntu-latest

    env:
      DBT_PROFILES_DIR: .
      DBT_SNOWFLAKE_ACCOUNT: ${{ secrets.DBT_SNOWFLAKE_ACCOUNT }}
      DBT_SNOWFLAKE_USER: ${{ secrets.DBT_PROD_USER }}
      DBT_SNOWFLAKE_PASSWORD: ${{ secrets.DBT_PROD_PASSWORD }}
      DBT_SNOWFLAKE_ROLE: TRANSFORMER
      DBT_SNOWFLAKE_DATABASE: ANALYTICS
      DBT_SNOWFLAKE_WAREHOUSE: TRANSFORMING
      DBT_SNOWFLAKE_SCHEMA: analytics_prod

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dbt
        run: pip install dbt-core==1.7.4 dbt-snowflake==1.7.1

      - name: Install dbt packages
        run: dbt deps

      - name: dbt debug
        run: dbt debug --target prod

      - name: dbt seed
        run: dbt seed --target prod --full-refresh

      - name: dbt run (staging)
        run: dbt run --select staging.* --target prod

      - name: dbt run (marts)
        run: dbt run --select marts.* --target prod

      - name: dbt test
        run: dbt test --target prod

      - name: dbt source freshness
        run: dbt source freshness --target prod

      - name: Generate documentation
        run: dbt docs generate --target prod

      - name: Upload docs to S3
        run: |
          aws s3 sync target/ s3://your-dbt-docs-bucket/latest/ \
            --exclude "*" \
            --include "manifest.json" \
            --include "catalog.json" \
            --include "index.html"
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: Save manifest for state comparison
        run: |
          aws s3 cp target/manifest.json s3://your-state-bucket/prod/manifest.json
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: Notify on failure
        if: failure()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {
              "text": "🚨 dbt Production run failed!",
              "blocks": [
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "*dbt Production Deploy Failed*\n\nRun: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
                  }
                }
              ]
            }
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

---

Additional Resources

• dbt Documentation: https://docs.getdbt.com/
• dbt Discourse: https://discourse.getdbt.com/
• dbt Slack Community: https://www.getdbt.com/community/
• dbt Learn: https://courses.getdbt.com/
• dbt Package Hub: https://hub.getdbt.com/

---

Last Updated: October 2025 Skill Version: 1.0.0

dbt Data Transformation

A comprehensive Claude Code skill for mastering dbt (data build tool) for analytics engineering and data transformation.

Overview

This skill provides complete guidance for building modern data transformation pipelines using dbt. Whether you're migrating from legacy ETL, starting a new analytics project, or optimizing existing data workflows, this skill covers everything from basic model development to advanced production deployment strategies.

What You'll Learn

• Model Development: Build SQL transformations using refs, sources, and CTEs
• Materializations: Choose the right strategy (view, table, incremental, ephemeral)
• Testing: Implement comprehensive data quality testing
• Documentation: Create searchable, auto-generated data catalogs
• Incremental Models: Efficiently process large datasets
• Macros & Jinja: Write reusable, dynamic SQL
• Package Management: Leverage and create dbt packages
• Production Workflows: Deploy with CI/CD, orchestration, and monitoring

Why dbt?

dbt transforms the way analytics teams work by bringing software engineering best practices to data transformation:

Key Benefits

1. Version Control for SQL: Track changes, collaborate with teams, and review transformations 2. Automated Testing: Ensure data quality with built-in and custom tests 3. Documentation: Auto-generated docs with lineage graphs and searchable catalog 4. Modularity: Reusable SQL through refs and macros reduces repetition 5. Dependency Management: Automatic DAG building ensures correct execution order 6. Development Workflow: Separate dev/prod environments, CI/CD integration 7. Performance: Incremental models and optimizations for large datasets

The Modern Data Stack

Data Sources → EL Tool → Data Warehouse → dbt → BI Tool
(Apps, APIs)   (Fivetran) (Snowflake)    (T)   (Looker)

dbt handles the "T" (Transform) in ELT, running inside your data warehouse for maximum performance.

When to Use This Skill

Use dbt when you need to:

• Transform raw data into analytics-ready datasets
• Build dimensional models (facts and dimensions)
• Create reusable data transformation logic
• Test data quality automatically
• Document data models and business logic
• Handle large-scale incremental data processing
• Implement DataOps practices
• Migrate from stored procedures or ETL tools
• Enable self-service analytics

Project Structure

A well-organized dbt project follows this structure:

my_dbt_project/
├── dbt_project.yml           # Project configuration
├── packages.yml              # Package dependencies
├── profiles.yml              # Database connections (not in repo)
├── README.md                 # Project documentation
│
├── models/                   # SQL transformation models
│   ├── staging/              # 1:1 with source tables
│   │   ├── jaffle_shop/
│   │   │   ├── _jaffle_shop__sources.yml
│   │   │   ├── _jaffle_shop__models.yml
│   │   │   ├── stg_jaffle_shop__customers.sql
│   │   │   └── stg_jaffle_shop__orders.sql
│   │   └── stripe/
│   │       ├── _stripe__sources.yml
│   │       └── stg_stripe__payments.sql
│   │
│   ├── intermediate/         # Purpose-built transformations
│   │   └── int_orders_joined.sql
│   │
│   └── marts/               # Business-defined entities
│       ├── core/
│       │   ├── _core__models.yml
│       │   ├── dim_customers.sql
│       │   └── fct_orders.sql
│       └── marketing/
│           └── fct_customer_sessions.sql
│
├── tests/                   # Custom data tests
│   └── assert_positive_totals.sql
│
├── macros/                  # Reusable Jinja-SQL
│   ├── cents_to_dollars.sql
│   └── grant_permissions.sql
│
├── seeds/                   # CSV reference data
│   └── country_codes.csv
│
├── snapshots/               # SCD Type 2 captures
│   └── customers_snapshot.sql
│
├── analyses/                # Ad-hoc queries
│   └── revenue_analysis.sql
│
└── target/                  # Compiled artifacts (gitignored)
    ├── compiled/
    ├── run/
    └── manifest.json

Quick Start Guide

1. Installation

# Install dbt Core with your database adapter
pip install dbt-core dbt-snowflake  # or dbt-bigquery, dbt-redshift, etc.

# Verify installation
dbt --version

2. Initialize Project

# Create new dbt project
dbt init my_analytics_project

cd my_analytics_project

3. Configure Connection

Edit ~/.dbt/profiles.yml:

my_analytics_project:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: abc123.us-east-1
      user: your_username
      password: "{{ env_var('DBT_PASSWORD') }}"
      role: transformer
      database: analytics
      warehouse: transforming
      schema: dbt_dev
      threads: 4

    prod:
      type: snowflake
      account: abc123.us-east-1
      user: prod_user
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      role: transformer
      database: analytics
      warehouse: transforming
      schema: analytics_prod
      threads: 8

4. Test Connection

dbt debug

5. Create Your First Model

-- models/staging/stg_customers.sql

with source as (
    select * from {{ source('jaffle_shop', 'customers') }}
),

renamed as (
    select
        id as customer_id,
        first_name,
        last_name,
        first_name || ' ' || last_name as customer_name,
        email
    from source
)

select * from renamed

6. Define Source

# models/staging/sources.yml

version: 2

sources:
  - name: jaffle_shop
    database: raw
    schema: jaffle_shop
    tables:
      - name: customers
        description: Raw customer data
        columns:
          - name: id
            description: Primary key
            tests:
              - unique
              - not_null

7. Run Your Model

# Run all models
dbt run

# Run specific model
dbt run --select stg_customers

# Run with tests
dbt build

8. Test Your Data

dbt test

9. Generate Documentation

dbt docs generate
dbt docs serve

Visit http://localhost:8080 to view your data documentation.

Core Concepts

Models

Models are SELECT statements that define data transformations:

-- Every model is a SELECT statement
select
    order_id,
    customer_id,
    order_date,
    status
from {{ ref('stg_orders') }}

Materializations

Control how models are built in your warehouse:

• View (default): Virtual table, query runs on access
• Table: Physical table, full rebuild each run
• Incremental: Only processes new data
• Ephemeral: CTE interpolated into dependent models

{{ config(materialized='incremental') }}

select * from {{ source('events', 'page_views') }}

{% if is_incremental() %}
    where event_timestamp > (select max(event_timestamp) from {{ this }})
{% endif %}

Tests

Ensure data quality with tests:

# Schema tests in YAML
models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id

-- Custom data tests in SQL
-- tests/assert_positive_totals.sql
select * from {{ ref('fct_orders') }}
where order_total < 0

Documentation

Document your models for discoverability:

models:
  - name: fct_orders
    description: |
      Order fact table containing one row per order.

      **Grain:** One row per order
      **Refresh:** Daily at 2 AM UTC

    columns:
      - name: order_id
        description: Primary key for orders
      - name: order_total
        description: Total order amount in USD

Macros

Reusable Jinja-SQL functions:

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, precision=2) %}
    round({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}

Usage:

select
    order_id,
    {{ cents_to_dollars('amount_cents') }} as amount_dollars
from {{ ref('stg_orders') }}

Common Use Cases

Use Case 1: E-commerce Analytics

Build a complete analytics pipeline for an e-commerce business:

Sources (Raw Data)
├── Application DB: customers, orders, order_items, products
└── Payment Provider: payments, refunds

Staging (Cleaned & Renamed)
├── stg_customers
├── stg_orders
├── stg_order_items
├── stg_products
└── stg_payments

Marts (Business Entities)
├── dim_customers (with lifetime value)
├── dim_products (with inventory)
├── fct_orders (order facts)
└── fct_order_items (line-level facts)

Metrics & Analytics
├── daily_revenue_metrics
├── customer_cohort_analysis
└── product_performance

Use Case 2: SaaS Product Analytics

Track user behavior and subscription metrics:

Event Tracking
├── stg_page_views
├── stg_feature_usage
└── stg_api_calls

User & Account Management
├── dim_users
├── dim_accounts
└── fct_subscriptions

Product Metrics
├── feature_adoption_rates
├── user_retention_cohorts
└── account_health_scores

Use Case 3: Marketing Attribution

Attribute revenue to marketing channels:

Marketing Data
├── stg_ad_clicks (Google, Facebook)
├── stg_email_opens
└── stg_referrals

Customer Journey
├── int_customer_touchpoints
└── int_attribution_windows

Attribution Models
├── first_touch_attribution
├── last_touch_attribution
└── multi_touch_attribution

Development Workflow

Daily Development

1. Pull latest changes: git pull origin main 2. Install dependencies: dbt deps 3. Create feature branch: git checkout -b feature/new-metric 4. Develop models: Write SQL in models/ 5. Run models: dbt run --select +my_new_model 6. Test: dbt test --select my_new_model 7. Document: Add descriptions in YAML 8. Commit & push: Git workflow 9. Open PR: Code review process

CI/CD Integration

• On PR: Run modified models and tests
• On merge to main: Full production deployment
• Scheduled: Daily/hourly production runs
• Monitoring: Track test failures, run times

Best Practices

Model Organization

1. Use the staging → intermediate → marts pattern 2. Keep staging models 1:1 with source tables 3. Name models clearly (stg_, int_, fct_, dim_) 4. One model per business concept

SQL Style

1. Use CTEs for readability 2. Lowercase SQL keywords 3. Consistent indentation (2 or 4 spaces) 4. Comment complex business logic

Testing

1. Test all primary keys (unique + not_null) 2. Test foreign key relationships 3. Add custom tests for business rules 4. Use appropriate severity levels

Documentation

1. Document model purpose and grain 2. Explain complex transformations 3. Keep documentation current 4. Link to external resources

Performance

1. Use incremental models for large datasets 2. Partition and cluster tables 3. Filter early in CTEs 4. Monitor query costs

Advanced Topics

Incremental Strategies

• Append: Add new rows only
• Merge: Upsert based on unique key
• Delete+Insert: Full partition replacement

Snapshots (SCD Type 2)

Track historical changes:

{% snapshot customers_snapshot %}
    {{
        config(
            target_schema='snapshots',
            unique_key='customer_id',
            strategy='timestamp',
            updated_at='updated_at'
        )
    }}
    select * from {{ source('app', 'customers') }}
{% endsnapshot %}

Exposures

Track downstream dependencies:

exposures:
  - name: executive_dashboard
    type: dashboard
    url: https://looker.company.com/dashboards/123
    depends_on:
      - ref('fct_orders')
      - ref('dim_customers')

Cross-Database Macros

Write database-agnostic SQL:

{% macro datediff(start_date, end_date, datepart) %}
    {{ adapter.dispatch('datediff')(start_date, end_date, datepart) }}
{% endmacro %}

Troubleshooting

Common Issues

"Compilation Error: Model not found"

• Check model name in ref()
• Ensure model file exists
• Run dbt compile to check for syntax errors

"Database Error: Relation does not exist"

• Check source configuration
• Verify database/schema names
• Run dbt run on upstream models first

"Incremental model running full refresh every time"

• Check is_incremental() logic
• Verify unique_key is set
• Ensure table exists (first run is full)

"Tests failing unexpectedly"

• Review test logic
• Check for data changes
• Use dbt test --select test_name to debug

Resources & Learning

Official Resources

• Documentation: https://docs.getdbt.com/
• Courses: https://courses.getdbt.com/
• Community: https://discourse.getdbt.com/
• Package Hub: https://hub.getdbt.com/

Community Packages

• dbt-utils: Essential utility macros
• dbt-expectations: Great Expectations-style tests
• audit-helper: Compare datasets
• codegen: Generate boilerplate code

Learning Path

1. Beginner: Complete dbt Fundamentals course 2. Intermediate: Build a complete project (staging → marts) 3. Advanced: Implement incremental models, macros, packages 4. Expert: CI/CD, custom materializations, performance tuning

Getting Help

• Documentation: https://docs.getdbt.com/
• Community Forum: https://discourse.getdbt.com/
• Slack: https://www.getdbt.com/community/
• GitHub Issues: https://github.com/dbt-labs/dbt-core/issues
• Stack Overflow: Tag questions with dbt

Skill Contents

This skill includes:

• SKILL.md: Comprehensive 20KB+ guide with 20+ detailed examples
• README.md: This overview and quick start guide
• EXAMPLES.md: 18+ practical examples with full code
• Context7 Integration: Real-world code snippets from dbt-core repository

---

Version: 1.0.0 Author: Claude Code Skills License: MIT Last Updated: October 2025