第17章 dbt数据转换¶

-- models/staging/stg_orders.sql
-- dbt自动创建表或视图，无需写CREATE TABLE

SELECT
    id AS order_id,
    user_id AS customer_id,
    order_date,
    status,
    amount AS order_amount
FROM {{ source('raw', 'orders') }}
WHERE order_date IS NOT NULL

-- models/staging/stg_orders.sql

WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),

renamed AS (
    SELECT
        id AS order_id,
        user_id AS customer_id,
        CAST(order_date AS DATE) AS order_date,
        CAST(amount AS DECIMAL(10, 2)) AS order_amount,
        LOWER(status) AS order_status,
        created_at,
        updated_at
    FROM source
    WHERE id IS NOT NULL  -- 过滤脏数据
)

SELECT * FROM renamed

-- models/intermediate/int_order_items_enriched.sql

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

order_items AS (
    SELECT * FROM {{ ref('stg_order_items') }}
),

products AS (
    SELECT * FROM {{ ref('stg_products') }}
)

SELECT
    oi.order_item_id,
    oi.order_id,
    o.customer_id,
    o.order_date,
    p.product_name,
    p.category,
    oi.quantity,
    oi.unit_price,
    oi.quantity * oi.unit_price AS line_total
FROM order_items oi
JOIN orders o ON oi.order_id = o.order_id
JOIN products p ON oi.product_id = p.product_id

-- models/marts/core/fct_orders.sql
{{ config(materialized='incremental', unique_key='order_id') }}

WITH order_items AS (
    SELECT * FROM {{ ref('int_order_items_enriched') }}
)

SELECT
    order_id,
    customer_id,
    order_date,
    COUNT(DISTINCT order_item_id) AS item_count,
    SUM(line_total) AS order_total,
    MIN(line_total) AS min_item_total,
    MAX(line_total) AS max_item_total
FROM order_items

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

GROUP BY order_id, customer_id, order_date

-- models/marts/core/dim_customers.sql

WITH customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
),

orders AS (
    SELECT * FROM {{ ref('fct_orders') }}
),

customer_orders AS (
    SELECT
        customer_id,
        MIN(order_date) AS first_order_date,
        MAX(order_date) AS most_recent_order_date,
        COUNT(order_id) AS total_orders,
        SUM(order_total) AS lifetime_value
    FROM orders
    GROUP BY customer_id
)

SELECT
    c.customer_id,
    c.customer_name,
    c.email,
    c.signup_date,
    co.first_order_date,
    co.most_recent_order_date,
    COALESCE(co.total_orders, 0) AS total_orders,
    COALESCE(co.lifetime_value, 0) AS lifetime_value,
    CASE
        WHEN co.lifetime_value >= 1000 THEN 'high'
        WHEN co.lifetime_value >= 100 THEN 'medium'
        ELSE 'low'
    END AS customer_tier
FROM customers c
LEFT JOIN customer_orders co ON c.customer_id = co.customer_id

-- ref()建立模型间的依赖关系，dbt自动构建DAG

-- 引用同项目model
SELECT * FROM {{ ref('stg_orders') }}

-- 引用其他dbt包的model
SELECT * FROM {{ ref('dbt_utils', 'date_spine') }}

-- source()引用源表
SELECT * FROM {{ source('raw', 'orders') }}

-- tests/assert_positive_order_amount.sql
-- 返回行数 > 0 即测试失败

SELECT
    order_id,
    order_amount
FROM {{ ref('fct_orders') }}
WHERE order_amount <= 0

-- tests/generic/test_is_positive.sql
{% test is_positive(model, column_name) %}

SELECT *
FROM {{ model }}
WHERE {{ column_name }} < 0

{% endtest %}

-- Jinja在dbt中的用法

-- 变量
{% set payment_methods = ['credit_card', 'bank_transfer', 'gift_card'] %}

-- 循环
SELECT
    order_id,
    {% for method in payment_methods %}
    SUM(CASE WHEN payment_method = '{{ method }}' THEN amount ELSE 0 END)
        AS {{ method }}_amount
    {% if not loop.last %},{% endif %}
    {% endfor %}
FROM {{ ref('stg_payments') }}
GROUP BY order_id

-- 条件
SELECT *
FROM {{ ref('fct_orders') }}
{% if target.name == 'dev' %}
WHERE order_date >= CURRENT_DATE - INTERVAL '30 days'
{% endif %}

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, decimal_places=2) %}
    ROUND({{ column_name }} / 100.0, {{ decimal_places }})
{% endmacro %}

-- 使用
SELECT
    order_id,
    {{ cents_to_dollars('amount_cents') }} AS amount_dollars
FROM {{ ref('stg_payments') }}

-- macros/generate_schema_name.sql（自定义schema命名规则）
{% 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 }}
    {%- endif -%}
{%- endmacro %}

-- 安装：packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: ">=1.0.0"

-- 使用surrogate_key
SELECT
    {{ dbt_utils.generate_surrogate_key(['order_id', 'product_id']) }} AS order_item_key,
    *
FROM {{ ref('stg_order_items') }}

-- Date Spine（生成日期维度）
{{ dbt_utils.date_spine(
    datepart="day",
    start_date="cast('2020-01-01' as date)",
    end_date="cast('2026-01-01' as date)"
) }}

-- Pivot
SELECT
    customer_id,
    {{ dbt_utils.pivot(
        'order_status',
        dbt_utils.get_column_values(ref('stg_orders'), 'order_status')
    ) }}
FROM {{ ref('stg_orders') }}
GROUP BY customer_id

-- 推荐：使用CTE让逻辑清晰

-- 导入CTE（引用依赖）
WITH orders AS (  -- CTE公共表表达式：临时命名结果集
    SELECT * FROM {{ ref('stg_orders') }}
),

customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
),

-- 逻辑CTE（业务转换）
customer_orders AS (
    SELECT
        customer_id,
        COUNT(*) AS order_count,
        SUM(order_amount) AS total_spent
    FROM orders
    GROUP BY customer_id  -- GROUP BY分组；HAVING过滤分组
),

-- 最终CTE（组装结果）
final AS (
    SELECT
        c.customer_id,
        c.customer_name,
        COALESCE(co.order_count, 0) AS order_count,
        COALESCE(co.total_spent, 0) AS total_spent
    FROM customers c
    LEFT JOIN customer_orders co USING (customer_id)  -- JOIN连接多个表
)

SELECT * FROM final

-- 策略一：Append（追加，适合事件型数据）
{{ config(
    materialized='incremental',
    incremental_strategy='append'
) }}

SELECT * FROM {{ ref('stg_events') }}
{% if is_incremental() %}
WHERE event_time > (SELECT MAX(event_time) FROM {{ this }})
{% endif %}

-- 策略二：Merge（合并，适合缓慢变化数据）
{{ config(
    materialized='incremental',
    unique_key='user_id',
    incremental_strategy='merge'
) }}

SELECT * FROM {{ ref('stg_users') }}
{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

{{ config(
    materialized='incremental',
    unique_key='event_id'
) }}

SELECT * FROM {{ ref('stg_events') }}
{% if is_incremental() %}
-- 回看3天数据处理延迟到达
WHERE event_time > (SELECT MAX(event_time) - INTERVAL '3 days' FROM {{ this }})  -- 子查询：嵌套在另一个查询中
{% endif %}