09. dbt 数据转换与 DataOps

-- models/staging/stg_orders.sql
{{ config(
    materialized='table',
    unique_key='order_id',
    partition_by={
      "field": "created_at",
      "data_type": "timestamp",
      "granularity": "day"
    }
) }}

WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),
renamed AS (
    SELECT
        order_id,
        user_id,
        CAST(amount AS DECIMAL(18,2)) AS amount,
        status,
        created_at,
        updated_at,
        -- 派生字段
        CASE 
            WHEN amount >= 1000 THEN 'high'
            WHEN amount >= 100 THEN 'medium'
            ELSE 'low'
        END AS value_tier
    FROM source
    WHERE created_at >= '2020-01-01'
)

SELECT * FROM renamed

-- models/marts/fct_orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',  -- merge / insert_overwrite / append
    on_schema_change='sync_all_columns'
) }}

SELECT
    order_id,
    user_id,
    amount,
    status,
    created_at
FROM {{ ref('stg_orders') }}

{% if is_incremental() %}
    -- 增量条件：只处理新数据
    WHERE created_at > (SELECT MAX(created_at) FROM {{ this }})
{% endif %}

-- 运行方式：
-- dbt run --full-refresh    → 全量重建（DELETE + 全量 INSERT）
-- dbt run                   → 增量追加（只处理新增数据）

-- {{ source() }} 引用上游原始数据
{{ source('raw_schema', 'orders') }}
-- 在 models/sources.yml 中定义：
-- sources:
--   - name: raw_schema
--     tables:
--       - name: orders
--         loaded_at_field: created_at

-- {{ ref() }} 引用 dbt 模型
{{ ref('stg_orders') }}
-- dbt 自动解析依赖关系，构建 DAG

# models/sources.yml
version: 2

sources:
  - name: raw_ecommerce
    database: raw_db
    schema: ecommerce
    tables:
      - name: orders
        description: "原始订单数据"
        columns:
          - name: order_id
            description: "订单唯一标识"
            tests:
              - not_null
              - unique
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
        loaded_at_field: created_at

      - name: users
        description: "原始用户数据"

# models/staging/staging.yml
version: 2

models:
  - name: stg_orders
    description: "清洗后的订单数据"
    columns:
      - name: order_id
        description: "主键"
        tests:
          - unique
          - not_null

      - name: user_id
        description: "用户ID，关联 dim_users"
        tests:
          - not_null
          - relationships:
              to: ref('stg_users')
              field: user_id

      - name: status
        description: "订单状态"
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'shipped', 'delivered', 'cancelled']

      - name: amount
        description: "订单金额"
        tests:
          - not_null
          - dbt_utils.greater_than_or_equal_to:
              value: 0

-- tests/assert_positive_revenue.sql
-- 单文件测试：检查总收入为正
SELECT
    SUM(amount) AS total_revenue
FROM {{ ref('fct_orders') }}
HAVING SUM(amount) < 0

-- 如果查询返回结果，则测试失败

-- macros/test_greater_than_or_equal_to.sql
{% test greater_than_or_equal_to(model, column_name, value) %}

SELECT
    {{ column_name }}
FROM {{ model }}
WHERE {{ column_name }} < {{ value }}

{% endtest %}

# 使用自定义测试
models:
  - name: stg_orders
    columns:
      - name: amount
        tests:
          - greater_than_or_equal_to:
              value: 0

-- 变量
{% set start_date = '2024-01-01' %}
WHERE dt >= '{{ start_date }}'

-- 逻辑控制
{% if target.name == 'prod' %}
    {{ config(materialized='table') }}
{% else %}
    {{ config(materialized='view') }}
{% endif %}

-- 循环
{% set payment_methods = ['credit_card', 'coupon', 'bank_transfer', 'gift_card'] %}

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

-- 生成：
-- SELECT order_id,
--   SUM(CASE WHEN payment_method = 'credit_card' THEN amount ELSE 0 END) AS credit_card_amount,
--   SUM(CASE WHEN payment_method = 'coupon' THEN amount ELSE 0 END) AS coupon_amount,
--   ...

-- macros/generate_surrogate_key.sql
{% macro generate_surrogate_key(field_list) %}
    -- 生成代理键（类似 dbt_utils 的 generate_surrogate_key）
    CAST(
        MD5(
            CONCAT(
                {% for field in field_list %}
                    COALESCE(CAST({{ field }} AS STRING), '_dbt_null_')
                    {%- if not loop.last -%}, '||', {% endif %}
                {% endfor %}
            )
        ) AS STRING
    )
{% endmacro %}

-- 使用
SELECT
    {{ generate_surrogate_key(['user_id', 'order_date']) }} AS sk_order,
    *
FROM {{ ref('stg_orders') }}

-- macros/date_spine.sql
{% macro date_spine(datepart, start_date, end_date) %}
    WITH date_series AS (
        SELECT {{ dbt.dateadd(datepart, "n", "'" ~ start_date ~ "'") }} AS date_{{ datepart }}
        FROM UNNEST(GENERATE_ARRAY(0, 
            DATE_DIFF('{{ end_date }}', '{{ start_date }}', {{ datepart }})
        )) AS n
    )
    SELECT * FROM date_series
{% endmacro %}

# models/docs.md
{% docs orders_status %}

订单状态说明：

| 状态 | 说明 |
|------|------|
| pending | 待支付 |
| paid | 已支付 |
| shipped | 已发货 |
| delivered | 已送达 |
| cancelled | 已取消 |

{% enddocs %}

# models/staging/staging.yml
version: 2
models:
  - name: stg_orders
    description: "订单明细表，来源 raw.orders"
    columns:
      - name: status
        description: '{{ doc("orders_status") }}'

# 生成文档
# dbt docs generate

# 启动文档服务器
# dbt docs serve --port 8080

# 产出：
# - index.html: 搜索和导航
# - Lineage DAG: 表血缘关系图
# - Table details: 列定义、测试、描述

# .github/workflows/dbt-ci.yml
name: dbt CI

on:
  pull_request:
    branches: [main]

jobs:
  dbt-ci:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dbt
        run: |
          pip install dbt-snowflake  # 或 dbt-bigquery, dbt-doris 等
          dbt deps
      
      - name: Compile
        run: dbt compile --target ci
      
      - name: Lint
        run: sqlfluff lint models/  # SQL 风格检查
      
      - name: Run tests on modified models
        run: |
          dbt run --select state:modified+ --target ci
          dbt test --select state:modified+ --target ci
        env:
          DBT_PROFILES_DIR: ./
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

      - name: Generate docs
        run: dbt docs generate
      
      - name: Upload docs artifact
        uses: actions/upload-artifact@v4
        with:
          name: dbt-docs
          path: target/

# 利用 dbt 的状态比较，只跑变更的模型
# dbt run --select state:modified+
# state:modified  = 相对于上一次 manifest 有变更的模型
# +               = 加上下游依赖

# 完整 Slim CI 流程
# 1. 下载 main 分支的 manifest.json
# 2. dbt compile --state path/to/main/manifest
# 3. dbt run --select state:modified+ --defer --state path/to/main/manifest
# 4. dbt test --select state:modified+

# ~/.dbt/profiles.yml
ecommerce:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: xyz.eu-west-1
      user: dbt_dev
      password: "{{ env_var('DBT_DEV_PASSWORD') }}"
      role: analyst
      database: analytics_dev
      warehouse: dev_wh
      schema: "{{ env_var('DBT_USER') }}"
      threads: 4

    prod:
      type: snowflake
      account: xyz.eu-west-1
      user: dbt_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      role: dbt_prod
      database: analytics
      warehouse: prod_wh
      schema: marts
      threads: 8

-- 1. 分区与聚簇
{{ config(
    materialized='incremental',
    partition_by={
        "field": "created_at",
        "data_type": "timestamp",
        "granularity": "day"
    },
    cluster_by=["user_id", "status"]
) }}

-- 2. 查询标签追踪
{{ config(
    query_tag = 'dbt_model_' ~ this.name
) }}

-- 3. 预钩子和后钩子
{{ config(
    pre_hook="ALTER SESSION SET QUERY_TAG = 'dbt_{{this.name}}'",
    post_hook="GRANT SELECT ON {{ this }} TO ROLE analyst"
) }}

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1

  - package: calogica/dbt_expectations
    version: 0.10.3

  - package: dbt-labs/codegen
    version: 0.12.1

  - git: "https://github.com/my-org/dbt-common-macros.git"
    revision: v1.2.0

# 安装依赖包
$ dbt deps

# seeds/country_codes.csv
country_code,country_name,region,currency
US,United States,North America,USD
CN,China,Asia,CNY
DE,Germany,Europe,EUR
JP,Japan,Asia,JPY
GB,United Kingdom,Europe,GBP

-- 在模型中引用 seed
SELECT
    o.*,
    c.country_name,
    c.currency
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('country_codes') }} c
    ON o.country_code = c.country_code

-- snapshots/snapshot_users.sql
{% snapshot snapshot_users %}

{{
    config(
      target_database='analytics',
      target_schema='snapshots',
      unique_key='user_id',
      strategy='timestamp',
      updated_at='updated_at',
    )
}}

SELECT * FROM {{ source('raw_ecommerce', 'users') }}

{% endsnapshot %}

-- merge 策略（Snowflake/BigQuery/Databricks）
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    merge_update_columns=['status', 'amount', 'updated_at']
) }}

SELECT * FROM {{ source('raw_ecommerce', 'orders') }}
{% if is_incremental() %}
    WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

-- insert_overwrite 策略（适合 BigQuery 分区表）
{{ config(
    materialized='incremental',
    partition_by={"field": "created_at", "data_type": "date"},
    incremental_strategy='insert_overwrite'
) }}

-- 每次运行覆盖最近 7 天分区的全量数据
SELECT * FROM {{ source('raw_ecommerce', 'orders') }}
WHERE created_at >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)

-- delete+insert 策略（Dremio/StarRocks）
{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='delete+insert'
) }}

# models/marts/marts.yml
version: 2

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null

      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'paid', 'shipped', 'delivered', 'cancelled']

      - name: user_id
        tests:
          - relationships:
              to: ref('dim_users')
              field: user_id

      - name: amount
        tests:
          - dbt_utils.greater_than_or_equal_to:
              value: 0

-- macros/test_column_is_date.sql
{% test column_is_date(model, column_name) %}

SELECT
    {{ column_name }}
FROM {{ model }}
WHERE SAFE.PARSE_DATE('%Y-%m-%d', CAST({{ column_name }} AS STRING)) IS NULL
  AND {{ column_name }} IS NOT NULL

{% endtest %}

-- tests/assert_payment_method_total.sql
SELECT
    payment_method,
    SUM(amount) AS total_amount
FROM {{ ref('fct_orders') }}
GROUP BY payment_method
HAVING total_amount < 0

# packages.yml
packages:
  - package: calogica/dbt_expectations
    version: 0.10.3

# 使用 dbt_expectations 测试
models:
  - name: fct_orders
    columns:
      - name: amount
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 100000

      - name: created_at
        tests:
          - dbt_expectations.expect_row_values_to_have_recent_data:
              datepart: day
              interval: 1

# 生成静态文档到 target/ 目录
$ dbt docs generate

# 本地预览文档
$ dbt docs serve --port 8080

# models/exposures.yml
version: 2

exposures:
  - name: executive_dashboard
    type: dashboard
    maturity: high
    url: https://looker.company.com/dashboards/123
    description: "高管日报仪表板，核心经营指标汇总"
    depends_on:
      - ref('fct_orders')
      - ref('fct_revenue_daily')
    owner:
      name: "数据产品团队"
      email: "data-product@company.com"

# metrics/revenue_metrics.yml （dbt Metrics v1 / MetricFlow v2）
version: 2

metrics:
  - name: total_revenue
    label: "总营收"
    model: ref('fct_orders')
    description: "所有已完成订单的金额总和"
    calculation_method: sum
    expression: amount
    timestamp: created_at
    time_grains: [day, week, month, quarter, year]
    dimensions:
      - status
      - country_code
    filters:
      - field: status
        operator: '!='
        value: 'cancelled'

# airflow/dags/dbt_pipeline.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

with DAG(
    dag_id='dbt_daily_pipeline',
    start_date=days_ago(1),
    schedule_interval='0 3 * * *',
    catchup=False
) as dag:

    dbt_deps = BashOperator(
        task_id='dbt_deps',
        bash_command='cd /opt/dbt && dbt deps --profiles-dir .'
    )

    dbt_run = BashOperator(
        task_id='dbt_run',
        bash_command='cd /opt/dbt && dbt run --target prod --profiles-dir .'
    )

    dbt_test = BashOperator(
        task_id='dbt_test',
        bash_command='cd /opt/dbt && dbt test --target prod --profiles-dir .'
    )

    dbt_deps >> dbt_run >> dbt_test

# 通过 dbt Cloud API 触发 Job
import requests

account_id = 12345
job_id = 67890
token = "dbtu_xxxxxxxx"

response = requests.post(
    f"https://cloud.getdbt.com/api/v2/accounts/{account_id}/jobs/{job_id}/run/",
    headers={"Authorization": f"Token {token}"},
    json={"cause": "Triggered by Airflow"}
)

# Airflow 中使用 SimpleHttpOperator
from airflow.providers.http.operators.http import SimpleHttpOperator

trigger_dbt_cloud_job = SimpleHttpOperator(
    task_id='trigger_dbt_cloud_job',
    http_conn_id='dbt_cloud_api',
    endpoint='api/v2/accounts/{{ var.value.dbt_account_id }}/jobs/{{ var.value.dbt_job_id }}/run/',
    method='POST',
    headers={"Content-Type": "application/json"},
    data='{"cause": "Airflow triggered"}'
)

# airflow/dags/cosmos_dag.py
from cosmos import DbtDag, ProjectConfig, ProfileConfig
from cosmos.profiles import SnowflakeUserPasswordProfileMapping

profile_config = ProfileConfig(
    profile_name="ecommerce",
    target_name="prod",
    profile_mapping=SnowflakeUserPasswordProfileMapping(
        conn_id="snowflake_conn",
        profile_args={"database": "analytics", "schema": "marts"}
    )
)

dbt_snowflake_dag = DbtDag(
    project_config=ProjectConfig("/opt/dbt/ecommerce_analytics"),
    profile_config=profile_config,
    dag_id="cosmos_dbt_pipeline",
    schedule_interval="0 3 * * *",
    start_date=days_ago(1),
    catchup=False,
    default_args={"retries": 2}
)

# ~/.dbt/profiles.yml
my_project:
  target: dev
  outputs:
    dev:
      type: bigquery
      method: oauth
      project: my-gcp-project
      dataset: "dbt_{{ env_var('USER') }}"   # 每人独立 schema
      threads: 4

    staging:
      type: bigquery
      method: service-account
      project: my-gcp-project
      dataset: staging
      threads: 8

    prod:
      type: bigquery
      method: service-account
      project: my-gcp-project
      dataset: analytics
      threads: 16

-- BigQuery 分区 + 聚簇优化
{{ config(
    materialized='incremental',
    partition_by={
        "field": "created_at",
        "data_type": "timestamp",
        "granularity": "day"
    },
    cluster_by=["user_id", "status"]
) }}

SELECT
    order_id,
    user_id,
    amount,
    status,
    created_at
FROM {{ source('raw_ecommerce', 'orders') }}
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)

# dbt_project.yml
models:
  my_project:
    +pre-hook:
      - "SET query_tag = 'dbt_{{ this.name }}'"
    +post-hook:
      - "GRANT SELECT ON {{ this }} TO ROLE analyst"
    staging:
      +snowflake_warehouse: "DEV_WH"
      +threads: 4
    marts:
      +snowflake_warehouse: "PROD_WH"
      +threads: 8

# 调整单命令并行度
$ dbt run --threads 16              # 默认 4，可提升至数据仓库上限
$ dbt run --select marts.fct_orders --threads 8

# profiles.yml
spark_datalake:
  target: dev
  outputs:
    dev:
      type: spark
      method: thrift
      host: spark-thrift-server.hadoop
      port: 10000
      schema: analytics
      threads: 4

-- 使用 Delta Lake 增量模型
{{ config(
    materialized='incremental',
    file_format='delta',
    incremental_strategy='merge',
    unique_key='order_id'
) }}

SELECT * FROM {{ source('bronze', 'orders') }}
{% if is_incremental() %}
    WHERE modified_at > (SELECT MAX(modified_at) FROM {{ this }})
{% endif %}

# profiles.yml
local_analytics:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: '/tmp/analytics.duckdb'
      threads: 4

# 本地极速运行，无需网络连接
$ dbt run --target dev --select staging
$ dbt test --target dev

# profiles.yml
federation:
  target: prod
  outputs:
    prod:
      type: trino
      host: trino.company.com
      port: 8080
      user: dbt
      catalog: iceberg
      schema: marts
      threads: 8