1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# models/mart/core_metrics.yml
models:
  - name: final_daily_sales
    config:
      tags: ["core", "daily"] # 命綱モデルにタグを貼る！

# 実行コマンド:
# coreタグがついたモデルのテストだけ動かす
dbt test --select tag:core 

# 変更したモデルとその影響範囲だけテストする
dbt test --select state:modified+

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- int_層のモデルの例:
-- stgから整形済みのデータを持ってきて、ビジネス上の重要な計算（ここでは初回購入日）をバッチリやる
SELECT
  u.user_id,
  u.user_name,
  MIN(o.order_date) AS first_purchase_date,
  COUNT(o.order_id) AS total_orders
FROM {{ ref('stg_ecommerce__users') }} AS u
LEFT JOIN {{ ref('stg_ecommerce__orders') }} AS o
  ON u.user_id = o.user_id
GROUP BY 1, 2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

-- incremental モデルの基本構造
{{ config(
    materialized='incremental', 
    unique_key='transaction_id', # データが重複しないようにキーを指定
    # DW側でパフォーマンスを上げる設定 (例: BigQueryのパーティション)
    partition_by={
        "field": "created_at",
        "data_type": "timestamp"
    }
) }}

SELECT * FROM {{ source('raw_data', 'transactions') }}
{% if is_incremental() %}
  -- ココが重要！前回動かした時以降のデータだけを対象にする
  WHERE created_at > (SELECT MAX(created_at) FROM {{ this }})
{% endif %}

1
2
3
4
5
6

# dbt_project.yml で設定
models:
  +post_hook: "ANALYZE TABLE {{ this }} COMPUTE STATISTICS" # 例: 実行後の最適化
  +on_schema_change: fail # ここを「fail」に！スキーマが変わったら、とりあえず止める（DROPさせない）

# profiles.yml で環境を分ける設定をして、本番（prod）では権限を絞る

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

-- 良い例: マクロを共通の型変換に利用
-- macros/timestamp_to_jst.sql
{% macro timestamp_to_jst(column_name) %}
  CONVERT_TZ({{ column_name }}, 'UTC', 'Asia/Tokyo')
{% endmacro %}

-- モデルでの使用:
SELECT
  user_id,
  {{ timestamp_to_jst('created_at') }} AS created_at_jst,
  ...
FROM {{ ref('stg_users') }}

1
2
3
4
5
6

# dbt_project.yml または models/config
models:
  # 本番のときだけincremental（増分）に、それ以外はtable（全構築）にする
  +materialized: "{% if target.name == 'prod' %}incremental{% else %}table{% endif %}"
  # ターゲット名に合わせてスキーマ名も自動で切り替える
  +schema: "{{ target.schema }}" 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

{% snapshot customers_snapshot %}

{{
    config(
      target_schema='snapshots',
      unique_key='customer_id', # 追跡したいデータのID
      strategy='check',         # 変更を検出する方法を指定
      check_cols=['first_name', 'email'], # 履歴を記録したいカラムを指定
      invalidate_hard_deletes=True
    )
}}

SELECT * FROM {{ source('raw_data', 'customers') }}

{% endsnapshot %}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# models/mart/final_daily_sales.yml
version: 2
models:
  - name: final_daily_sales
    description: "BIツールで使う、日次売上集計の最終形です。" # 必ず説明を書く！
    columns:
      - name: day_key
        description: "売上日のキー（YYYYMMDD形式）。"
        tests:
          - unique
          - not_null
      - name: total_sales_usd
        description: "その日の総売上（米ドル）。マイナスになることはないはず。"
        tests:
          - dbt_expectations.expect_column_to_be_between:
              min_value: 0
              max_value: 99999999

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# dbt_project.yml
seeds:
  ja_dbt_project:
    # dbt seedを動かす時、config_dataフォルダにあるCSVだけを対象にする
    +select: ["config_data"]
    +enabled: true

# 実行コマンド:
# 小さな設定データだけをサクッとロードする
dbt seed