dbt Explained — SQL-First Data Transformations with dbt Core

-- models/stg_orders.sql
-- Stage model: clean and standardize raw orders
with source as (
    select * from {{ source('ecommerce', 'orders') }}
),

cleaned as (
    select
        order_id,
        customer_id,
        cast(order_date as date) as order_date,
        cast(amount as decimal(10,2)) as amount,
        lower(status) as status,
        case
            when status = 'completed' then 'delivered'
            when status = 'shipped' then 'in_transit'
            else status
        end as delivery_status
    from source
    where order_id is not null
)

select * from cleaned

# models/sources.yml
version: 2

sources:
  - name: ecommerce
    database: raw_db
    schema: public
    tables:
      - name: orders
        description: "Raw orders from the e-commerce platform"
        columns:
          - name: order_id
            description: "Primary key"
            tests:
              - unique
              - not_null
      - name: customers
        description: "Raw customer records"
  - name: web_analytics
    database: raw_db
    schema: web
    tables:
      - name: page_views
        loaded_at_field: viewed_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}

-- tests/assert_positive_order_amount.sql
-- Custom test: fail if any order has negative amount
select * from {{ ref('stg_orders') }}
where amount < 0

-- models/stg_customers.sql
-- Stage: clean customer data
with source as (
    select * from {{ source('ecommerce', 'customers') }}
)

select
    customer_id,
    trim(full_name) as customer_name,
    lower(email) as email,
    city,
    state,
    country,
    signup_date,
    case
        when total_orders >= 10 then 'vip'
        when total_orders >= 1 then 'active'
        else 'new'
    end as customer_tier
from source

-- models/dim_customers.sql
-- Dimension: customer details
select
    customer_id,
    customer_name,
    email,
    city,
    state,
    country,
    customer_tier,
    datediff('day', signup_date, current_date) as tenure_days
from {{ ref('stg_customers') }}

-- models/fct_orders.sql
-- Fact: order transactions
with orders as (
    select * from {{ ref('stg_orders') }}
),
customers as (
    select * from {{ ref('dim_customers') }}
)

select
    o.order_id,
    o.customer_id,
    o.order_date,
    o.amount,
    o.delivery_status,
    c.customer_tier,
    c.country,
    date_trunc('month', o.order_date) as order_month
from orders o
left join customers c on o.customer_id = c.customer_id

# dbt_project.yml
name: dodatech_analytics
version: '1.0'
config-version: 2
profile: dodatech_profile

model-paths: ["models"]
test-paths: ["tests"]
analysis-paths: ["analyses"]

models:
  dodatech_analytics:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: analytics
      dims:
        +materialized: table
      facts:
        +materialized: incremental
        +unique_key: order_id

# dbt_demo.py
# Simulate dbt model compilation and execution
import json

class dbtModel:
    def __init__(self, name, sql, materialization='view', refs=None):
        self.name = name
        self.sql = sql
        self.materialization = materialization
        self.refs = refs or []

    def compile(self):
        """Simulate dbt's SQL compilation (ref resolution)."""
        compiled = self.sql
        for ref in self.refs:
            placeholder = "{{ ref('" + ref + "') }}"
            compiled = compiled.replace(placeholder, f"dodatech_analytics.{ref}")
        return compiled

    def execute(self, data):
        """Simulate dbt model execution."""
        print(f"[dbt] Running model: {self.name}")
        print(f"[dbt] Materialization: {self.materialization}")
        print(f"[dbt] Compiled SQL ({len(self.compile())} chars):")
        print(f"  {self.compile()[:100]}...")
        print(f"[dbt] Result: {len(data)} rows {'inserted' if self.materialization == 'incremental' else 'created'}")
        return data

# Define models programmatically
stg_orders_sql = """
select order_id, customer_id, order_date,
       cast(amount as decimal(10,2)) as amount, status
from raw_orders
where order_id is not null
"""

fct_orders_sql = """
select o.order_id, o.customer_id, o.order_date,
       o.amount, o.status, c.customer_tier
from {{ ref('stg_orders') }} o
left join {{ ref('dim_customers') }} c
on o.customer_id = c.customer_id
"""

models = {
    "stg_orders": dbtModel("stg_orders", stg_orders_sql, "view"),
    "dim_customers": dbtModel("dim_customers", "select * from raw_customers", "table"),
    "fct_orders": dbtModel("fct_orders", fct_orders_sql, "incremental", refs=["stg_orders", "dim_customers"]),
}

# Simulate dbt run
print("=== dbt Run ===\n")
sample_data = [{"order_id": i, "amount": i * 100.0} for i in range(1, 6)]

for name, model in models.items():
    model.execute(sample_data)

print("\n=== dbt Lineage ===")
print("""
  raw_orders ──> stg_orders ──┐
                              ├──> fct_orders
  raw_customers ─> dim_customers ┘
""")

# Run all models
dbt run

# Run specific model with dependencies
dbt run --model fct_orders+

# Run tests
dbt test

# Generate documentation
dbt docs generate
dbt docs serve  # Opens docs in browser

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

# Debug connection
dbt debug

# freshness_check.py
# Simulate dbt source freshness checking
from datetime import datetime, timedelta
import json

class SourceFreshness:
    def __init__(self, max_hours):
        self.max_hours = max_hours

    def check(self, source_name, table_name, last_loaded, current_time=None):
        if current_time is None:
            current_time = datetime.now()
        elapsed = (current_time - last_loaded).total_seconds() / 3600
        status = "PASS" if elapsed <= self.max_hours else "FAIL"
        return {
            "source": source_name,
            "table": table_name,
            "last_loaded": last_loaded.isoformat(),
            "elapsed_hours": round(elapsed, 1),
            "max_hours": self.max_hours,
            "status": status,
        }

checker = SourceFreshness(max_hours=24)

tables = [
    ("ecommerce", "orders", datetime.now() - timedelta(hours=2)),
    ("ecommerce", "customers", datetime.now() - timedelta(hours=18)),
    ("web_analytics", "page_views", datetime.now() - timedelta(hours=30)),
]

print(f"{'Source':<20} {'Table':<15} {'Last Loaded':<25} {'Elapsed':<10} {'Max':<10} {'Status'}")
print("-" * 90)
for source, table, last_loaded in tables:
    result = checker.check(source, table, last_loaded)
    print(f"{result['source']:<20} {result['table']:<15} {result['last_loaded']:<25} {result['elapsed_hours']:<10}h {result['max_hours']:<10}h {result['status']}")