dbt Integration

# Create and activate a virtual environment
python3 -m venv dbt-venv
source dbt-venv/bin/activate

# Install dbt-core and the Materialize adapter
pip install dbt-core dbt-materialize

dbt --version

materialize_project:
  target: dev
  outputs:
    dev:
      type: materialize
      host: instance.region.aws.materialize.cloud
      port: 6875
      user: [email protected]
      pass: mzp_your_app_password_here
      dbname: materialize
      schema: dbt_user
      cluster: quickstart
      sslmode: require
      threads: 1
      keepalives_idle: 0
      retries: 1

materialize_project:
  target: dev
  outputs:
    dev:
      type: materialize
      host: dev-instance.region.aws.materialize.cloud
      port: 6875
      user: [email protected]
      pass: "{{ env_var('DBT_MATERIALIZE_PASSWORD') }}"
      dbname: materialize
      schema: dbt_dev
      cluster: dev_cluster
      sslmode: require
      
    prod:
      type: materialize
      host: prod-instance.region.aws.materialize.cloud
      port: 6875
      user: [email protected]
      pass: "{{ env_var('DBT_MATERIALIZE_PASSWORD') }}"
      dbname: materialize
      schema: public
      cluster: prod_cluster
      sslmode: require

-- models/active_users.sql
{{ config(materialized='materialized_view') }}

SELECT 
    user_id,
    COUNT(*) as event_count,
    MAX(event_timestamp) as last_seen
FROM {{ source('app', 'events') }}
GROUP BY user_id

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

SELECT * FROM {{ ref('active_users') }}
WHERE event_count > 10

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

SELECT * FROM {{ ref('user_segments') }}

-- models/sources/kafka_events.sql
{{ config(
    materialized='source',
    source_type='kafka'
) }}

CREATE SOURCE kafka_events
FROM KAFKA CONNECTION kafka_connection (
    TOPIC 'events'
)
FORMAT JSON

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

CREATE SINK user_sink
FROM {{ ref('active_users') }}
INTO KAFKA CONNECTION kafka_connection (
    TOPIC 'active-users'
)
FORMAT JSON

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

SELECT * FROM events WHERE event_type = 'click'

{{ config(
    materialized='materialized_view',
    indexes=[
        {'columns': ['user_id']},
        {'columns': ['user_id', 'event_date'], 'name': 'user_date_idx'},
        {'default': True, 'cluster': 'analytics'}
    ]
) }}

SELECT 
    user_id,
    DATE(event_timestamp) as event_date,
    COUNT(*) as event_count
FROM {{ source('app', 'events') }}
GROUP BY user_id, event_date

-- Single column index
{{ config(indexes=[{'columns': ['user_id']}]) }}

-- Multi-column index with custom name
{{ config(indexes=[{'columns': ['user_id', 'timestamp'], 'name': 'user_time_idx'}]) }}

-- Default index (all columns)
{{ config(indexes=[{'default': True}]) }}

-- Index in specific cluster
{{ config(indexes=[{'columns': ['user_id'], 'cluster': 'analytics'}]) }}

# models/schema.yml
version: 2

models:
  - name: active_users
    description: "Users with recent activity"
    columns:
      - name: user_id
        description: "Unique user identifier"
        tests:
          - unique
          - not_null
      - name: event_count
        description: "Number of events"
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"

# Store failures as materialized views
dbt test --store-failures

# Store failures as views (recommended for Materialize)
dbt test --store-failures --store-failures-as view

tests:
  +store_failures: true
  +store_failures_as: view
  +schema: dbt_test_audit

# Generate documentation
dbt docs generate

# Serve documentation locally
dbt docs serve

models:
  my_project:
    +persist_docs:
      relation: true
      columns: true

{{ config(
    pre_hook="SET cluster = analytics_cluster",
    post_hook="GRANT SELECT ON {{ this }} TO analyst_role"
) }}

SELECT * FROM {{ ref('base_events') }}

{{ config(
    schema='analytics',
    materialized='materialized_view'
) }}

SELECT * FROM {{ ref('raw_events') }}

# packages.yml
packages:
  - package: materializeinc/materialize_dbt_utils
    version: 0.4.0

dbt deps

# Run all models
dbt run

# Run specific model
dbt run --select active_users

# Run models and downstream dependencies
dbt run --select active_users+

# Test models
dbt test

# Generate documentation
dbt docs generate

# Compile without running
dbt compile

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

{{ config(
    materialized='materialized_view',
    indexes=[{'columns': ['user_id', 'timestamp']}]
) }}

dev:
  cluster: dev_cluster
prod:
  cluster: prod_cluster

pass: "{{ env_var('DBT_MATERIALIZE_PASSWORD') }}"

dbt debug

SELECT mz_version();

SHOW CLUSTERS;
SHOW CLUSTER REPLICAS;

my_dbt_project/
├── dbt_project.yml
├── profiles.yml (in ~/.dbt/)
├── models/
│   ├── staging/
│   │   ├── stg_events.sql
│   │   └── stg_users.sql
│   ├── intermediate/
│   │   ├── int_user_events.sql
│   │   └── schema.yml
│   └── marts/
│       ├── active_users.sql
│       └── user_metrics.sql
├── tests/
└── macros/