dbt Core: models, refs, sources и materialization types (table, view, incremental)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as spark_sum, to_date, count

spark = SparkSession.builder.appName("User Metrics ETL").getOrCreate()

raw_events = spark.read.parquet("s3a://bronze/events/")

clean_events = raw_events \
    .filter(col("event_type").isNotNull()) \
    .filter(col("user_id") > 0) \
    .withColumn("event_date", to_date(col("event_ts"))) \
    .select("user_id", "session_id", "event_type", "event_date", "revenue")

clean_events.createOrReplaceTempView("stg_events")

user_metrics = spark.sql("""
    SELECT
        user_id,
        event_date,
        COUNT(DISTINCT session_id) AS sessions,
        SUM(revenue)               AS total_revenue,
        COUNT(*)                   AS event_count
    FROM stg_events
    GROUP BY user_id, event_date
""")

user_metrics.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("event_date") \
    .saveAsTable("gold.fct_user_activity")

-- models/gold/fct_user_activity.sql
{{ config(
    materialized='incremental',
    unique_key=['user_id', 'event_date'],
    partition_by={'field': 'event_date', 'data_type': 'date'}
) }}

SELECT
    user_id,
    event_date,
    COUNT(DISTINCT session_id) AS sessions,
    SUM(revenue)               AS total_revenue,
    COUNT(*)                   AS event_count
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_date >= (SELECT MAX(event_date) FROM {{ this }})
{% endif %}

GROUP BY user_id, event_date

pip install dbt-spark[PyHive]

dbt --version

my_dbt_project/
├── dbt_project.yml          # Главный конфиг проекта
├── profiles.yml             # Подключения к БД (обычно в ~/.dbt/)
├── models/                  # SQL-трансформации
│   ├── staging/             # stg_* - нормализация Bronze
│   │   ├── sources.yml      # Описание источников
│   │   ├── stg_events.sql
│   │   └── stg_users.sql
│   ├── intermediate/        # int_* - промежуточная логика
│   │   └── int_sessions.sql
│   └── marts/               # fct_*, dim_* - Gold-витрины
│       ├── fct_user_activity.sql
│       └── dim_campaigns.sql
├── macros/                  # Jinja-макросы
│   └── generate_schema.sql
├── tests/                   # Кастомные тесты
│   └── assert_revenue_positive.sql
├── seeds/                   # CSV-файлы (справочники)
│   └── campaign_types.csv
└── snapshots/               # SCD Type 2 (история изменений)
    └── snap_user_status.sql

name: 'analytics_lakehouse'
version: '1.0.0'
config-version: 2

profile: 'spark_local'

model-paths: ["models"]
test-paths: ["tests"]
macro-paths: ["macros"]
seed-paths: ["seeds"]
snapshot-paths: ["snapshots"]

models:
  analytics_lakehouse:
    staging:
      +materialized: view
      +schema: staging
    intermediate:
      +materialized: ephemeral
      +schema: intermediate
    marts:
      +materialized: table
      +schema: marts
      fct_user_activity:
        +materialized: incremental

spark_local:
  target: dev
  outputs:
    dev:
      type: spark
      method: thrift
      host: localhost
      port: 10000
      user: hadoop
      schema: dbt_dev
      connect_retries: 3
      connect_timeout: 10
      server_side_parameters:
        spark.sql.shuffle.partitions: "200"
        spark.sql.adaptive.enabled: "true"

    prod:
      type: spark
      method: thrift
      host: spark-thrift.internal
      port: 10000
      user: dbt_service
      schema: dbt_prod
      server_side_parameters:
        spark.sql.shuffle.partitions: "2000"
        spark.sql.adaptive.enabled: "true"
        spark.sql.adaptive.advisoryPartitionSizeInBytes: "134217728"

-- stg_events.sql - ПЛОХО
SELECT * FROM bronze.raw_events WHERE event_date = '2024-01-15'

# models/staging/sources.yml
version: 2

sources:
  - name: bronze
    description: "Bronze-слой: сырые данные от ingestion-пайплайнов"
    database: bronze_db
    schema: events
    tables:
      - name: raw_events
        description: "Сырые события пользователей из Kafka"
        identifier: raw_events
        loaded_at_field: event_ts
        freshness:
          warn_after: {count: 1, period: hour}
          error_after: {count: 3, period: hour}
        columns:
          - name: event_id
            description: "Уникальный ID события"
            tests:
              - not_null
              - unique
          - name: user_id
            description: "ID пользователя"
            tests:
              - not_null
          - name: event_type
            description: "Тип события"
          - name: event_ts
            description: "Временная метка события"
            tests:
              - not_null
          - name: revenue
            description: "Выручка (только для purchase событий)"

      - name: raw_users
        description: "Справочник пользователей из PostgreSQL CDC"
        identifier: raw_users
        loaded_at_field: updated_at
        freshness:
          warn_after: {count: 6, period: hour}
          error_after: {count: 24, period: hour}

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

SELECT
    event_id,
    user_id,
    session_id,
    event_type,
    CAST(event_ts AS TIMESTAMP)            AS event_ts,
    CAST(event_ts AS DATE)                 AS event_date,
    COALESCE(revenue, 0.0)                 AS revenue,
    LOWER(TRIM(device_type))               AS device_type,
    CURRENT_TIMESTAMP()                    AS dbt_loaded_at
FROM {{ source('bronze', 'raw_events') }}
WHERE event_id IS NOT NULL
  AND user_id IS NOT NULL

dbt source freshness

11:23:45  Running with dbt=1.7.0
11:23:45  Found 3 sources, 0 freshness errors, 0 freshness warnings
11:23:46  bronze.raw_events ... [PASS (23 minutes ago)]
11:23:46  bronze.raw_users  ... [WARN (4 hours ago)]

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

SELECT
    user_id,
    LOWER(email)                                AS email,
    COALESCE(first_name, 'Unknown')             AS first_name,
    COALESCE(last_name, 'Unknown')              AS last_name,
    CAST(created_at AS TIMESTAMP)               AS created_at,
    CAST(created_at AS DATE)                    AS signup_date,
    CASE
        WHEN account_type = 'premium' THEN TRUE
        WHEN account_type = 'trial'   THEN TRUE
        ELSE FALSE
    END                                         AS is_paid_user,
    CURRENT_TIMESTAMP()                         AS dbt_loaded_at
FROM {{ source('bronze', 'raw_users') }}
WHERE user_id IS NOT NULL
  AND email IS NOT NULL

{{ config(...) }}
SELECT ...
{% if condition %}
  WHERE ...
{% endif %}

SELECT
    '{{ env_var("DBT_ENV", "dev") }}'        AS environment,
    '{{ var("run_date", "2024-01-01") }}'    AS run_date,
    CURRENT_TIMESTAMP()                       AS compiled_at
FROM ...

-- models/intermediate/int_sessions.sql
{{ config(materialized='ephemeral') }}

SELECT
    user_id,
    session_id,
    MIN(event_ts)                                AS session_start,
    MAX(event_ts)                                AS session_end,
    MAX(event_ts) - MIN(event_ts)                AS session_duration,
    COUNT(*)                                     AS event_count,
    SUM(revenue)                                 AS session_revenue,
    COUNT(DISTINCT event_type)                   AS unique_event_types,
    CAST(MIN(event_ts) AS DATE)                  AS session_date
FROM {{ ref('stg_events') }}
GROUP BY user_id, session_id

-- models/marts/fct_user_activity.sql
{{ config(materialized='table') }}

SELECT
    e.user_id,
    u.email,
    u.is_paid_user,
    e.event_date,
    COUNT(DISTINCT e.session_id)    AS sessions,
    SUM(s.session_duration)         AS total_session_time,
    SUM(e.revenue)                  AS total_revenue,
    COUNT(*)                        AS event_count
FROM {{ ref('stg_events') }}       AS e
JOIN {{ ref('stg_users') }}        AS u USING (user_id)
JOIN {{ ref('int_sessions') }}     AS s USING (user_id, session_id)
GROUP BY e.user_id, u.email, u.is_paid_user, e.event_date

sources/bronze/raw_events → stg_events ─┬─→ int_sessions ─┐
                                         │                  ↓
sources/bronze/raw_users  → stg_users ──┴────────────→ fct_user_activity

dbt compile --select fct_user_activity

-- target/compiled/analytics_lakehouse/models/marts/fct_user_activity.sql
SELECT
    e.user_id,
    u.email,
    ...
FROM dbt_dev.stg_events AS e
JOIN dbt_dev.stg_users  AS u USING (user_id)
JOIN dbt_dev.int_sessions AS s USING (user_id, session_id)
GROUP BY e.user_id, u.email, u.is_paid_user, e.event_date

dbt docs generate
dbt docs serve

CREATE OR REPLACE VIEW dbt_dev.stg_events AS
SELECT
    event_id,
    user_id,
    ...
FROM bronze_db.events.raw_events
WHERE event_id IS NOT NULL

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

SELECT
    event_id,
    user_id,
    CAST(event_ts AS TIMESTAMP) AS event_ts,
    CAST(event_ts AS DATE)      AS event_date,
    event_type,
    COALESCE(revenue, 0.0)      AS revenue
FROM {{ source('bronze', 'raw_events') }}
WHERE event_id IS NOT NULL

-- models/intermediate/int_sessions.sql
{{ config(materialized='ephemeral') }}

SELECT
    user_id,
    session_id,
    MIN(event_ts) AS session_start,
    MAX(event_ts) AS session_end,
    COUNT(*)      AS event_count
FROM {{ ref('stg_events') }}
GROUP BY user_id, session_id

-- Скомпилированный fct_user_activity.sql
WITH int_sessions AS (
    SELECT
        user_id,
        session_id,
        MIN(event_ts) AS session_start,
        MAX(event_ts) AS session_end,
        COUNT(*)      AS event_count
    FROM dbt_dev.stg_events
    GROUP BY user_id, session_id
)
SELECT
    e.user_id,
    ...
FROM dbt_dev.stg_events AS e
JOIN int_sessions        AS s USING (user_id, session_id)

-- Что генерирует dbt при table materialization
CREATE TABLE dbt_dev.fct_revenue_summary AS
SELECT
    campaign_id,
    channel,
    event_date,
    SUM(revenue) AS total_revenue,
    COUNT(*)     AS conversions
FROM dbt_dev.stg_events
GROUP BY campaign_id, channel, event_date

{{ config(
    materialized='table',
    file_format='delta',
    partition_by=['event_date'],
    clustered_by=['campaign_id'],
    buckets=8,
    location='s3a://datalake/gold/fct_revenue_summary'
) }}

SELECT
    campaign_id,
    channel,
    event_date,
    SUM(revenue) AS total_revenue,
    COUNT(*)     AS conversions
FROM {{ ref('stg_events') }}
GROUP BY campaign_id, channel, event_date

{{ config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge',
    file_format='delta',
    partition_by=['event_date'],
) }}

SELECT
    event_id,
    user_id,
    session_id,
    event_type,
    event_date,
    revenue,
    CURRENT_TIMESTAMP() AS processed_at
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_date >= (SELECT MAX(event_date) FROM {{ this }})
{% endif %}

{{ config(
    materialized='incremental',
    incremental_strategy='append',
    partition_by=['event_date'],
    file_format='parquet',
) }}

SELECT
    event_id,
    user_id,
    event_date,
    revenue
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_date > (SELECT MAX(event_date) FROM {{ this }})
{% endif %}

{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by=['event_date'],
    file_format='parquet',
) }}

SELECT
    event_id,
    user_id,
    event_date,
    channel,
    SUM(revenue) AS daily_revenue
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_date >= DATE_ADD(CURRENT_DATE(), -3)
{% endif %}

GROUP BY event_id, user_id, event_date, channel

server_side_parameters:
  spark.sql.sources.partitionOverwriteMode: "dynamic"

{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['user_id', 'event_date'],
    file_format='delta',
    partition_by=['event_date'],
) }}

SELECT
    user_id,
    event_date,
    COUNT(DISTINCT session_id) AS sessions,
    SUM(revenue)               AS total_revenue,
    COUNT(*)                   AS event_count,
    CURRENT_TIMESTAMP()        AS updated_at
FROM {{ ref('stg_events') }}

{% if is_incremental() %}
WHERE event_date >= DATE_ADD(CURRENT_DATE(), -7)
{% endif %}

GROUP BY user_id, event_date

MERGE INTO dbt_prod.fct_user_activity AS DBT_INTERNAL_DEST
USING (
    SELECT
        user_id,
        event_date,
        sessions,
        total_revenue,
        event_count,
        updated_at
    FROM dbt_prod.fct_user_activity__dbt_tmp
) AS DBT_INTERNAL_SOURCE
ON (
    DBT_INTERNAL_SOURCE.user_id = DBT_INTERNAL_DEST.user_id
    AND DBT_INTERNAL_SOURCE.event_date = DBT_INTERNAL_DEST.event_date
)
WHEN MATCHED THEN UPDATE SET
    sessions      = DBT_INTERNAL_SOURCE.sessions,
    total_revenue = DBT_INTERNAL_SOURCE.total_revenue,
    event_count   = DBT_INTERNAL_SOURCE.event_count,
    updated_at    = DBT_INTERNAL_SOURCE.updated_at
WHEN NOT MATCHED THEN INSERT (
    user_id, event_date, sessions, total_revenue, event_count, updated_at
) VALUES (
    DBT_INTERNAL_SOURCE.user_id,
    DBT_INTERNAL_SOURCE.event_date,
    DBT_INTERNAL_SOURCE.sessions,
    DBT_INTERNAL_SOURCE.total_revenue,
    DBT_INTERNAL_SOURCE.event_count,
    DBT_INTERNAL_SOURCE.updated_at
)

dbt run --select fct_user_activity --full-refresh

# models/staging/schema.yml
version: 2

models:
  - name: stg_events
    description: "Нормализованные события из Bronze"
    columns:
      - name: event_id
        description: "Уникальный ID события"
        tests:
          - not_null
          - unique
      - name: user_id
        description: "ID пользователя"
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id
      - name: event_type
        tests:
          - not_null
          - accepted_values:
              values: ['click', 'view', 'purchase', 'signup']
      - name: revenue
        tests:
          - not_null

  - name: fct_user_activity
    columns:
      - name: user_id
        tests:
          - not_null
      - name: event_date
        tests:
          - not_null
      - name: total_revenue
        tests:
          - not_null

SELECT COUNT(*) AS failures
FROM dbt_dev.stg_events
WHERE event_id IS NULL

-- tests/assert_revenue_non_negative.sql
SELECT
    user_id,
    event_date,
    total_revenue
FROM {{ ref('fct_user_activity') }}
WHERE total_revenue < 0

dbt test --select fct_user_activity

-- macros/add_audit_columns.sql
{% macro add_audit_columns() %}
    CURRENT_TIMESTAMP()                 AS created_at,
    CURRENT_TIMESTAMP()                 AS updated_at,
    '{{ env_var("DBT_RUNNER", "dbt") }}'  AS dbt_run_by,
    '{{ invocation_id }}'              AS dbt_invocation_id
{% endmacro %}

SELECT
    user_id,
    event_date,
    SUM(revenue) AS total_revenue,
    {{ add_audit_columns() }}
FROM {{ ref('stg_events') }}
GROUP BY user_id, event_date

-- macros/safe_divide.sql
{% macro safe_divide(numerator, denominator, default=0) %}
    CASE WHEN {{ denominator }} = 0
         THEN {{ default }}
         ELSE {{ numerator }} / {{ denominator }}
    END
{% endmacro %}

SELECT
    campaign_id,
    clicks,
    conversions,
    {{ safe_divide('conversions', 'clicks') }} AS conversion_rate
FROM {{ ref('fct_campaign_metrics') }}

pip install dbt-spark[PyHive] dbt-core

dbt init analytics_lakehouse

cd analytics_lakehouse

# ~/.dbt/profiles.yml
analytics_lakehouse:
  target: dev
  outputs:
    dev:
      type: spark
      method: session
      schema: dbt_dev
      server_side_parameters:
        spark.sql.shuffle.partitions: "50"
        spark.sql.adaptive.enabled: "true"
        spark.sql.extensions: "io.delta.sql.DeltaSparkSessionExtension"
        spark.sql.catalog.spark_catalog: "org.apache.spark.sql.delta.catalog.DeltaCatalog"

# models/staging/sources.yml
version: 2

sources:
  - name: bronze
    description: "Bronze слой - сырые данные"
    database: bronze
    schema: raw
    tables:
      - name: events
        identifier: events
        description: "Сырые события пользователей"
        loaded_at_field: event_ts
        freshness:
          warn_after: {count: 2, period: hour}
          error_after: {count: 6, period: hour}
        columns:
          - name: event_id
            tests: [not_null, unique]
          - name: user_id
            tests: [not_null]
          - name: event_ts
            tests: [not_null]

      - name: users
        identifier: users
        description: "Справочник пользователей из CDC"
        loaded_at_field: updated_at

-- models/staging/stg_events.sql
{{ config(
    materialized='view',
    tags=['staging', 'events']
) }}

SELECT
    event_id,
    CAST(user_id AS BIGINT)                 AS user_id,
    session_id,
    event_type,
    CAST(event_ts AS TIMESTAMP)             AS event_ts,
    CAST(event_ts AS DATE)                  AS event_date,
    COALESCE(CAST(revenue AS DECIMAL(12,4)), 0.0) AS revenue,
    LOWER(TRIM(COALESCE(channel, 'unknown'))) AS channel,
    LOWER(TRIM(COALESCE(device_type, 'unknown'))) AS device_type,
    CURRENT_TIMESTAMP()                     AS dbt_loaded_at
FROM {{ source('bronze', 'events') }}
WHERE event_id IS NOT NULL
  AND user_id  IS NOT NULL
  AND event_ts IS NOT NULL

-- models/staging/stg_users.sql
{{ config(
    materialized='view',
    tags=['staging', 'users']
) }}

SELECT
    CAST(user_id AS BIGINT)                     AS user_id,
    LOWER(TRIM(email))                          AS email,
    COALESCE(first_name, 'Unknown')             AS first_name,
    COALESCE(last_name, 'Unknown')              AS last_name,
    CAST(created_at AS TIMESTAMP)               AS created_at,
    CAST(created_at AS DATE)                    AS signup_date,
    CASE
        WHEN account_type IN ('premium', 'enterprise') THEN TRUE
        ELSE FALSE
    END                                         AS is_paid_user,
    COALESCE(country_code, 'XX')                AS country_code,
    CURRENT_TIMESTAMP()                         AS dbt_loaded_at
FROM {{ source('bronze', 'users') }}
WHERE user_id IS NOT NULL
  AND email   IS NOT NULL

-- models/intermediate/int_user_sessions.sql
{{ config(materialized='ephemeral') }}

SELECT
    user_id,
    session_id,
    event_date,
    MIN(event_ts)                                   AS session_start,
    MAX(event_ts)                                   AS session_end,
    DATEDIFF(SECOND, MIN(event_ts), MAX(event_ts))  AS session_duration_sec,
    COUNT(*)                                        AS events_in_session,
    SUM(revenue)                                    AS session_revenue,
    COUNT(DISTINCT event_type)                      AS unique_event_types,
    MAX(channel)                                    AS channel,
    MAX(device_type)                                AS device_type
FROM {{ ref('stg_events') }}
GROUP BY user_id, session_id, event_date

-- models/marts/fct_user_activity.sql
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['user_id', 'event_date'],
    file_format='delta',
    partition_by=['event_date'],
    tags=['gold', 'daily', 'user-metrics']
) }}

WITH sessions AS (
    SELECT * FROM {{ ref('int_user_sessions') }}

    {% if is_incremental() %}
    WHERE event_date >= DATE_ADD(CURRENT_DATE(), -3)
    {% endif %}
),

events AS (
    SELECT * FROM {{ ref('stg_events') }}

    {% if is_incremental() %}
    WHERE event_date >= DATE_ADD(CURRENT_DATE(), -3)
    {% endif %}
),

user_info AS (
    SELECT user_id, email, is_paid_user, country_code, signup_date
    FROM {{ ref('stg_users') }}
)

SELECT
    e.user_id,
    u.email,
    u.is_paid_user,
    u.country_code,
    u.signup_date,
    e.event_date,

    COUNT(DISTINCT e.session_id)        AS sessions,
    SUM(s.session_duration_sec)         AS total_session_sec,
    AVG(s.session_duration_sec)         AS avg_session_sec,
    SUM(e.revenue)                      AS total_revenue,
    COUNT(*)                            AS event_count,
    COUNT(DISTINCT e.event_type)        AS unique_event_types,

    COUNT(DISTINCT s.channel)           AS channels_used,
    MAX(s.channel)                      AS primary_channel,

    CURRENT_TIMESTAMP()                 AS updated_at

FROM events e
LEFT JOIN user_info u ON e.user_id = u.user_id
LEFT JOIN sessions  s ON e.user_id = s.user_id
                      AND e.session_id = s.session_id
                      AND e.event_date = s.event_date
GROUP BY
    e.user_id, u.email, u.is_paid_user, u.country_code, u.signup_date, e.event_date

# models/marts/schema.yml
version: 2

models:
  - name: fct_user_activity
    description: |
      Дневные метрики активности пользователей.
      Обновляется инкрементально через MERGE по (user_id, event_date).
      Окно обновления: последние 3 дня (для компенсации late data).
    columns:
      - name: user_id
        description: "ID пользователя"
        tests: [not_null]
      - name: event_date
        description: "Дата события"
        tests: [not_null]
      - name: total_revenue
        description: "Суммарная выручка за день"
        tests: [not_null]
      - name: sessions
        tests: [not_null]

dbt run --select staging.* marts.fct_user_activity

dbt run --select fct_user_activity+

dbt run --select +fct_user_activity+

dbt run --select fct_user_activity --full-refresh

dbt test --select fct_user_activity

cat target/compiled/analytics_lakehouse/models/marts/fct_user_activity.sql

cat logs/dbt.log | grep "MERGE INTO"

dbt compile --select fct_user_activity

cat target/compiled/analytics_lakehouse/models/marts/fct_user_activity.sql | \
  spark-sql --master spark://localhost:7077 -e "EXPLAIN FORMAT=FORMATTED $(cat -)"

dbt docs generate

dbt docs serve --port 8080

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

git checkout main
dbt compile
cp -r target/ target_prod/

git checkout feature/new-metric
dbt run --select state:modified+ \
         --state target_prod/

-- models/marts/everything.sql  ← ПЛОХО
SELECT
    e.event_id,
    e.user_id,
    u.email,
    u.country,
    c.campaign_name,
    c.budget,
    p.product_name,
    p.category,
    s.session_duration,
    ...
    (500 строк SQL)
FROM bronze.raw_events e
LEFT JOIN bronze.raw_users u ON e.user_id = u.user_id
LEFT JOIN bronze.raw_campaigns c ON e.campaign_id = c.campaign_id
LEFT JOIN bronze.raw_products p ON e.product_id = p.product_id
LEFT JOIN ...

-- ПЛОХО: хардкод путей
SELECT * FROM prod_db.gold.fct_user_activity
JOIN bronze.raw.events ON ...

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

SELECT
    event_date,
    SUM(revenue) AS total_revenue
FROM {{ source('bronze', 'events') }}
GROUP BY event_date

-- models/marts/fct_revenue.sql
{{ config(materialized='incremental') }}

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

{% if is_incremental() %}
-- Забыли добавить WHERE условие!
{% endif %}

{{ config(
    materialized='incremental',
    unique_key='user_id',
    incremental_strategy='merge'
) }}

SELECT
    user_id,
    event_date,
    SUM(revenue) AS total_revenue
FROM {{ ref('stg_events') }}
GROUP BY user_id, event_date

SELECT /*+ BROADCAST(u) */
    e.user_id,
    u.email,
    SUM(e.revenue) AS total_revenue
FROM {{ ref('stg_events') }} e
JOIN {{ ref('stg_users') }} u ON e.user_id = u.user_id
GROUP BY e.user_id, u.email

-- models/marts/fct_sales_daily.sql
{{ config(materialized='table') }}

SELECT
    order_id,
    user_id,
    product_id,
    order_date,
    CAST(quantity AS INT)           AS quantity,
    CAST(unit_price AS DECIMAL(12,4)) AS unit_price,
    quantity * unit_price           AS line_revenue,
    status,
    updated_at
FROM {{ ref('stg_orders') }}
WHERE status IN ('completed', 'shipped')