Transforming Data with dbt: Patterns That Actually Work in Production

Layer 1: Staging           Layer 2: Intermediate       Layer 3: Marts
------------------         -------------------         -----------------
One model per source       Cross-source joins           Business-domain tables
Clean and rename           Business logic               Aggregations
Cast types                 Derived metrics              Report-ready
No aggregation             Intermediate calcs           Used by BI tools

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

cleaned as (
    select
        customer_id,
        lower(trim(email))                         as email,
        initcap(trim(first_name))                  as first_name,
        initcap(trim(last_name))                   as last_name,
        initcap(trim(first_name))
            || ' ' || initcap(trim(last_name))     as full_name,
        cast(created_at as date)                   as signup_date,
        coalesce(lower(trim(status)), 'unknown')   as customer_status
    from source
    where customer_id is not null
      and email is not null
      and email like '%@%'
),

deduplicated as (
    select *,
        row_number() over (
            partition by email
            order by signup_date asc
        ) as row_num
    from cleaned
)

select * from deduplicated
where row_num = 1

with orders as (
    select * from {{ ref('stg_orders') }}
    where status = 'completed'
),

products as (
    select * from {{ ref('stg_products') }}
),

joined as (
    select
        o.order_id,
        o.order_date,
        o.customer_id,
        o.order_amount_usd,
        p.product_category,
        p.product_line
    from orders o
    left join products p on o.product_id = p.product_id
),

monthly as (
    select
        date_trunc('month', order_date)   as revenue_month,
        product_category,
        product_line,
        count(distinct order_id)          as order_count,
        count(distinct customer_id)       as unique_customers,
        sum(order_amount_usd)             as total_revenue_usd,
        avg(order_amount_usd)             as avg_order_value_usd
    from joined
    group by 1, 2, 3
)

select * from monthly
order by revenue_month, product_category

with touchpoints as (
    select * from {{ ref('stg_marketing_touchpoints') }}
),

conversions as (
    select
        customer_id,
        min(order_date) as first_purchase_date
    from {{ ref('stg_orders') }}
    where status = 'completed'
    group by 1
),

pre_purchase_touchpoints as (
    select
        t.customer_id,
        t.channel,
        t.event_time,
        t.campaign_name,
        c.first_purchase_date,
        datediff('day', t.event_time, c.first_purchase_date) as days_before_purchase
    from touchpoints t
    inner join conversions c
        on t.customer_id = c.customer_id
    where t.event_time <= c.first_purchase_date
      and datediff('day', t.event_time, c.first_purchase_date) <= 30
),

channel_summary as (
    select
        customer_id,
        channel,
        campaign_name,
        count(*) as touchpoint_count,
        min(event_time) as first_touch,
        max(event_time) as last_touch
    from pre_purchase_touchpoints
    group by 1, 2, 3
)

select * from channel_summary

with transactions as (
    select * from {{ ref('stg_transactions') }}
),

customer_baselines as (
    select
        customer_id,
        avg(amount_usd)   as avg_transaction_amount,
        stddev(amount_usd) as stddev_transaction_amount,
        count(*)           as transaction_count
    from transactions
    where transaction_date >= dateadd('month', -3, current_date)
    group by 1
),

flagged as (
    select
        t.transaction_id,
        t.customer_id,
        t.amount_usd,
        t.transaction_date,
        t.merchant_category,
        b.avg_transaction_amount,
        b.transaction_count,
        case
            when b.transaction_count < 5
                then 'insufficient_history'
            when t.amount_usd > (b.avg_transaction_amount + 3 * b.stddev_transaction_amount)
                then 'statistical_outlier'
            when t.amount_usd > 10000
                then 'high_value_threshold'
            else 'normal'
        end as risk_flag
    from transactions t
    left join customer_baselines b on t.customer_id = b.customer_id
)

select * from flagged
where risk_flag != 'normal'

with shipments as (
    select * from {{ ref('stg_shipments') }}
),

performance as (
    select
        shipment_id,
        order_id,
        carrier_name,
        origin_warehouse,
        destination_country,
        promised_delivery_date,
        actual_delivery_date,
        datediff('day', promised_delivery_date, actual_delivery_date) as days_delta,
        case
            when actual_delivery_date is null
                then 'in_transit'
            when actual_delivery_date <= promised_delivery_date
                then 'on_time'
            when datediff('day', promised_delivery_date, actual_delivery_date) <= 2
                then 'slightly_late'
            else 'significantly_late'
        end as delivery_status
    from shipments
)

select * from performance

version: 2

models:
  - name: fct_monthly_revenue
    columns:
      - name: revenue_month
        tests:
          - not_null
      - name: total_revenue_usd
        tests:
          - not_null

  - name: fct_delivery_performance
    columns:
      - name: shipment_id
        tests:
          - unique
          - not_null
      - name: delivery_status
        tests:
          - accepted_values:
              values: ['on_time', 'slightly_late', 'significantly_late', 'in_transit']

-- tests/assert_valid_risk_flags.sql
select transaction_id
from {{ ref('fct_transaction_risk_flags') }}
where risk_flag not in ('statistical_outlier', 'high_value_threshold', 'insufficient_history')