SAS Data Management Migration: SAS DI to ADF + dbt¶

-- models/staging/stg_clean_orders.sql
{{ config(materialized='view') }}

SELECT
    order_id,
    order_date,
    COALESCE(ship_date, DATE_ADD(order_date, 7)) AS ship_date,
    YEAR(order_date) AS order_year,
    QUARTER(order_date) AS order_quarter,
    amount * exchange_rate AS amount_usd,
    CASE
        WHEN country IN ('US', 'CA', 'MX') THEN 'Americas'
        WHEN country IN ('GB', 'DE', 'FR', 'IT', 'ES') THEN 'EMEA'
        ELSE 'APAC'
    END AS region,
    customer_id,
    product_id,
    country,
    order_date AS _loaded_at
FROM {{ source('oracle_prod', 'orders') }}
WHERE order_date >= '2025-01-01'
  AND status != 'CANCELLED'

-- models/staging/stg_claims_clean.sql (Step 1)
{{ config(materialized='view') }}

SELECT
    claim_id,
    DATE(claim_datetime) AS claim_date,
    TRIM(provider_id) AS provider_id,
    UPPER(diagnosis) AS diagnosis,
    amount,
    patient_id
FROM {{ source('claims_raw', 'claims') }}

-- models/staging/stg_claims_enriched.sql (Step 2)
{{ config(materialized='view') }}

SELECT
    c.*,
    p.provider_name,
    p.specialty
FROM {{ ref('stg_claims_clean') }} c
LEFT JOIN {{ ref('seed_providers') }} p
    ON c.provider_id = p.provider_id

-- models/gold/fact_daily_claims.sql (Step 3)
{{ config(
    materialized='incremental',
    unique_key=['provider_id', 'claim_date'],
    incremental_strategy='merge'
) }}

SELECT
    provider_id,
    provider_name,
    specialty,
    claim_date,
    SUM(amount) AS total_amount,
    COUNT(*) AS claim_count,
    AVG(amount) AS avg_amount
FROM {{ ref('stg_claims_enriched') }}
{% if is_incremental() %}
WHERE claim_date >= DATE_SUB(CURRENT_DATE(), 3)
{% endif %}
GROUP BY provider_id, provider_name, specialty, claim_date

-- models/intermediate/int_running_balance.sql
SELECT
    account_id,
    transaction_date,
    amount,
    SUM(amount) OVER (
        PARTITION BY account_id
        ORDER BY transaction_date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS running_total
FROM {{ ref('stg_transactions') }}

-- models/intermediate/int_enriched_orders.sql
SELECT
    o.*,
    c.customer_name,
    c.segment,
    c.region,
    CASE WHEN c.customer_id IS NOT NULL THEN 'Y' ELSE 'N' END AS match_flag
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('stg_customers') }} c
    ON o.customer_id = c.customer_id

-- macros/apply_age_group.sql
{% macro age_group(age_column) %}
  CASE
    WHEN {{ age_column }} < 18 THEN 'Minor'
    WHEN {{ age_column }} < 35 THEN 'Young Adult'
    WHEN {{ age_column }} < 55 THEN 'Middle Age'
    WHEN {{ age_column }} < 65 THEN 'Pre-Retirement'
    ELSE 'Senior'
  END
{% endmacro %}

-- models/staging/stg_patients.sql
SELECT
    patient_id,
    age,
    {{ age_group('age') }} AS age_group,
    state_code,
    s.state_name,
    risk_code,
    CASE
        WHEN risk_code IN ('A', 'B') THEN 'Low Risk'
        WHEN risk_code IN ('C', 'D') THEN 'Medium Risk'
        WHEN risk_code IN ('E', 'F') THEN 'High Risk'
        ELSE 'Unknown'
    END AS risk_category
FROM {{ source('raw', 'patients') }} p
LEFT JOIN {{ ref('state_lookup') }} s
    ON p.state_code = s.state_code

-- Validate row counts match between SAS and Delta
SELECT
    'sas_original' AS source,
    COUNT(*) AS row_count,
    COUNT(DISTINCT customer_id) AS distinct_customers,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount
FROM bronze.sas_migrated.clean_orders

UNION ALL

SELECT
    'dbt_model' AS source,
    COUNT(*) AS row_count,
    COUNT(DISTINCT customer_id) AS distinct_customers,
    SUM(amount_usd) AS total_amount,
    AVG(amount_usd) AS avg_amount
FROM gold.analytics.fact_orders