dbt

dbt は何をしますか？

dbtを使用すると、アナリティクスエンジニアはSQLのSELECT文を記述してデータを変換できます。マテリアライゼーション（テーブル、ビュー、インクリメンタル）、テスト、ドキュメント、リネージ追跡を処理します。

インストール

# PostgreSQLアダプターでdbtをインストール
pip install dbt-postgres

# または他のアダプターで
pip install dbt-bigquery
pip install dbt-snowflake

# 新しいプロジェクトを初期化
dbt init my_project
cd my_project

プロジェクト構造

my_project/
├── dbt_project.yml      # プロジェクト設定
├── profiles.yml         # 接続プロファイル（通常は~/.dbt/）
├── models/
│   ├── staging/         # 生データのクリーニング
│   │   ├── _staging.yml # スキーマ + ステージングモデルのテスト
│   │   ├── stg_users.sql
│   │   └── stg_orders.sql
│   └── marts/           # ビジネスロジック
│       ├── _marts.yml
│       └── fct_revenue.sql
├── tests/               # カスタムデータテスト
├── macros/              # 再利用可能なSQLマクロ
└── seeds/               # 読み込むCSVファイル

設定

# dbt_project.yml: プロジェクト設定
name: my_project
version: '1.0.0'
profile: my_project

models:
  my_project:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: analytics

# profiles.yml: データベース接続 (~/.dbt/profiles.yml)
my_project:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: analyst
      password: "{{ env_var('DBT_PASSWORD') }}"
      dbname: analytics
      schema: dev
      threads: 4
    prod:
      type: postgres
      host: prod-db.example.com
      port: 5432
      user: dbt_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      dbname: analytics
      schema: public
      threads: 8

ステージングモデル

-- models/staging/stg_users.sql: 生ユーザーデータをクリーニング
WITH source AS (
    SELECT * FROM {{ source('raw', 'users') }}
),

cleaned AS (
    SELECT
        id AS user_id,
        LOWER(TRIM(email)) AS email,
        name,
        created_at::timestamp AS signed_up_at,
        CASE WHEN status = 'active' THEN TRUE ELSE FALSE END AS is_active
    FROM source
    WHERE email IS NOT NULL
)

SELECT * FROM cleaned

-- models/staging/stg_orders.sql: 生注文データをクリーニング
SELECT
    id AS order_id,
    user_id,
    amount_cents / 100.0 AS amount,
    status,
    created_at::timestamp AS ordered_at
FROM {{ source('raw', 'orders') }}
WHERE status != 'test'

マートモデル

-- models/marts/fct_revenue.sql: 収益ファクトテーブル
{{
    config(
        materialized='incremental',
        unique_key='order_date',
        on_schema_change='sync_all_columns'
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
    WHERE ordered_at > (SELECT MAX(order_date) FROM {{ this }})
    {% endif %}
),

daily AS (
    SELECT
        DATE_TRUNC('day', ordered_at)::date AS order_date,
        COUNT(*) AS total_orders,
        COUNT(DISTINCT user_id) AS unique_customers,
        SUM(amount) AS total_revenue,
        AVG(amount) AS avg_order_value
    FROM orders
    WHERE status = 'completed'
    GROUP BY 1
)

SELECT * FROM daily

スキーマとテスト

# models/staging/_staging.yml: ソース、カラム、テストを定義
version: 2

sources:
  - name: raw
    schema: public
    tables:
      - name: users
        loaded_at_field: created_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
      - name: orders

models:
  - name: stg_users
    description: クリーニング済みユーザーデータ
    columns:
      - name: user_id
        tests: [unique, not_null]
      - name: email
        tests: [unique, not_null]

  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'completed', 'cancelled', 'refunded']

CLIコマンド

# commands.sh: 一般的なdbt CLIコマンド
# すべてのモデルを実行
dbt run

# 特定のモデルとその上流依存関係を実行
dbt run --select +fct_revenue

# テストを実行
dbt test

# ドキュメントを生成して提供
dbt docs generate
dbt docs serve --port 8081

# ソースの鮮度をチェック
dbt source freshness

# フルビルド（実行 + テスト + スナップショット）
dbt build

# プロダクションに対して実行
dbt run --target prod

マクロ

-- macros/cents_to_dollars.sql: 通貨変換のための再利用可能なマクロ
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)::numeric(10,2)
{% endmacro %}

-- モデルでの使用例: SELECT {{ cents_to_dollars('amount_cents') }} AS amount