Dbt Engineer

dbt_project/
  dbt_project.yml
  packages.yml
  models/
    staging/              # 1:1 with source tables
      _staging__sources.yml
      _staging__models.yml
      stg_salesforce__accounts.sql
      stg_stripe__charges.sql
    intermediate/         # Business logic transforms
      _int__models.yml
      int_opportunities_joined_to_accounts.sql
    marts/                # Final business entities
      finance/
        _finance__models.yml
        fct_monthly_revenue.sql
        dim_subscription.sql
  macros/
    generate_schema_name.sql
    cents_to_dollars.sql
  tests/
    generic/
      test_accepted_range.sql
    singular/
      assert_total_revenue_positive.sql
  seeds/
    country_codes.csv
  snapshots/
    snap_accounts.sql

name: 'company_analytics'
version: '1.0.0'
config-version: 2
profile: 'company_analytics'

models:
  company_analytics:
    staging:
      +materialized: view
      +schema: staging
    intermediate:
      +materialized: ephemeral
    marts:
      +materialized: table
      finance:
        +schema: finance
        +grants:
          select: ['finance_analyst_role']

-- Pattern 1: Append-only with watermark
{{
    config(
        materialized='incremental',
        unique_key='event_id',
        incremental_strategy='merge',
        on_schema_change='append_new_columns'
    )
}}

SELECT event_id, user_id, event_type, occurred_at, _loaded_at
FROM {{ ref('stg_segment__events') }}
{% if is_incremental() %}
WHERE _loaded_at > (SELECT MAX(_loaded_at) FROM {{ this }})
{% endif %}


-- Pattern 2: Late-arriving data with lookback window
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        incremental_strategy='merge'
    )
}}

SELECT order_id, customer_id, order_status, total_amount, updated_at
FROM {{ ref('stg_shopify__orders') }}
{% if is_incremental() %}
WHERE updated_at >= (SELECT DATEADD('day', -3, MAX(updated_at)) FROM {{ this }})
{% endif %}


-- Pattern 3: Insert-overwrite with partitioning
{{
    config(
        materialized='incremental',
        unique_key='surrogate_key',
        incremental_strategy='insert_overwrite',
        partition_by={'field': 'event_date', 'data_type': 'date', 'granularity': 'day'},
        cluster_by=['event_type', 'user_id']
    )
}}

SELECT
    {{ dbt_utils.generate_surrogate_key(['event_id', 'event_date']) }} AS surrogate_key,
    event_id, event_date, event_type, user_id
FROM {{ ref('stg_amplitude__events') }}
{% if is_incremental() %}
WHERE event_date >= _dbt_max_partition
{% endif %}

version: 2
models:
  - name: fct_monthly_revenue
    description: Monthly revenue by subscription and product line
    columns:
      - name: revenue_month
        data_tests:
          - not_null
          - dbt_utils.not_constant
      - name: subscription_id
        data_tests:
          - not_null
          - relationships:
              to: ref('dim_subscription')
              field: subscription_id
      - name: mrr_amount
        data_tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1000000
      - name: currency_code
        data_tests:
          - accepted_values:
              values: ['USD', 'EUR', 'GBP', 'CAD', 'AUD']

-- tests/generic/test_row_count_within_range.sql
{% test row_count_within_range(model, min_count, max_count) %}
WITH row_count AS (SELECT COUNT(*) AS cnt FROM {{ model }})
SELECT cnt FROM row_count
WHERE cnt < {{ min_count }} OR cnt > {{ max_count }}
{% endtest %}

-- tests/singular/assert_revenue_reconciles.sql
WITH source_total AS (
    SELECT SUM(amount_cents) / 100.0 AS total
    FROM {{ source('stripe', 'charges') }}
    WHERE status = 'succeeded' AND created >= '2024-01-01'
),
mart_total AS (
    SELECT SUM(charge_amount) AS total
    FROM {{ ref('fct_charges') }} WHERE charge_date >= '2024-01-01'
)
SELECT s.total AS source, m.total AS mart, ABS(s.total - m.total) AS diff
FROM source_total s CROSS JOIN mart_total m
WHERE ABS(s.total - m.total) > 1.00

-- macros/generate_schema_name.sql
{% macro generate_schema_name(custom_schema_name, node) %}
    {% set default_schema = target.schema %}
    {% if custom_schema_name is not none and target.name == 'prod' %}
        {{ custom_schema_name | trim }}
    {% else %}
        {{ default_schema }}_{{ custom_schema_name | trim }}
    {% endif %}
{% endmacro %}

-- macros/safe_divide.sql
{% macro safe_divide(numerator, denominator, default_value=0) %}
    CASE WHEN {{ denominator }} = 0 OR {{ denominator }} IS NULL
    THEN {{ default_value }}
    ELSE {{ numerator }}::FLOAT / {{ denominator }} END
{% endmacro %}

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, precision=2) %}
    ROUND({{ column_name }}::NUMERIC / 100, {{ precision }})
{% endmacro %}

# Source freshness configuration
sources:
  - name: salesforce
    database: raw
    schema: salesforce
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    loaded_at_field: _fivetran_synced
    tables:
      - name: account
        columns:
          - name: id
            data_tests: [unique, not_null]
      - name: opportunity
        freshness:
          error_after: {count: 6, period: hour}

# .github/workflows/dbt-ci.yml
name: dbt CI
on:
  pull_request:
    paths: ['models/**', 'macros/**', 'tests/**', 'dbt_project.yml']

jobs:
  dbt-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: install via pip: dbt-snowflake==1.7.*
      - run: dbt deps
      - run: dbt compile --target ci
      - run: |
          dbt run --select state:modified+ --defer --state ./prod-manifest
          dbt test --select state:modified+ --defer --state ./prod-manifest
      - run: sqlfluff lint models/ --dialect snowflake

# Enforce column types and prevent breaking changes
models:
  - name: fct_orders
    config:
      contract:
        enforced: true
    columns:
      - name: order_id
        data_type: varchar(36)
      - name: customer_id
        data_type: varchar(36)
      - name: order_date
        data_type: date
      - name: total_amount
        data_type: number(12,2)

[ ] Profile slow models with EXPLAIN / query plan
[ ] Convert large views to tables or incremental
[ ] Use ephemeral for models only referenced once
[ ] Partition by date column (BigQuery, Snowflake, Databricks)
[ ] Cluster by high-cardinality filter columns
[ ] Push filters early (filter in CTEs, not final SELECT)
[ ] Avoid SELECT * in production models
[ ] Replace correlated subqueries with JOINs
[ ] Use approximate functions for large aggregations