mz-dbt

dbt-materialize

dbt-materialize アダプターを使用すると、dbt のモデルベースのワークフローで Materialize のストリーミングパイプラインを管理できます。dbt-postgres を拡張し、Materialize 固有のマテリアライゼーション、ブルー／グリーンデプロイ用マクロ、クラスター管理機能を提供します。

このスキルの使い方

このスキルは単体で利用できます。ユーザーが dbt モデルを作成し、プロファイルを設定し、デプロイを実行して問題を解決するために必要な情報は、すべて以下に記載されています。dbt-materialize のソースコードや Materialize モノレポにアクセスする必要はありません。

1. ユーザーが Materialize で dbt をセットアップしたい場合：以下のプロファイル設定を参照してください。
2. ユーザーが特定のマテリアライゼーションについて質問している場合：マテリアライゼーションのセクションを参照してください。
3. ユーザーがブルー／グリーンデプロイを行いたい場合：ブルー／グリーンデプロイのセクションを参照してください。
4. ユーザーがインデックス、クラスター、設定について質問している場合：インデックス、Strict Mode、またはクラスター管理のセクションを参照してください。
5. ユーザーがエラーや予期しない動作に遭遇した場合：末尾の注意事項を参照してください。

アダプターの内部実装を詳しく調べる必要がある場合（通常は不要）、ソースは GitHub の MaterializeInc/materialize リポジトリ内の misc/dbt-materialize/ にあります。公式のユーザー向けドキュメントは https://materialize.com/docs/manage/dbt/ にあります。

インストール

pip install dbt-core dbt-materialize

Materialize v0.68.0 以降が必要です。

プロファイル設定

~/.dbt/profiles.yml 内で設定します。

my_project:
  target: dev
  outputs:
    dev:
      type: materialize
      host: <host>
      port: 6875
      user: <user>
      pass: <password>
      dbname: materialize
      schema: <dbt_schema>
      cluster: <cluster_name>       # optional, defaults to user's default cluster
      sslmode: require
      threads: 1                    # must be 1, Materialize does not support concurrent sessions

要点：

• threads は 1 にする必要があります。Materialize は複数の DDL セッションの同時実行をサポートしていません。
• cluster は省略可能ですが、設定することをおすすめします。省略すると、ユーザーのデフォルトクラスターが使用されます。
• オートコミットは常に有効です。Materialize はトランザクション内での任意のクエリ実行をサポートしていません。
• 本番環境のデプロイにはサービスアカウントを使用してください。

マテリアライゼーション

materialized_view（主なパターン）

ストリーミング変換で最も一般的なマテリアライゼーションです。上流のデータが変化すると自動で更新される、増分管理ビューを作成します。

-- models/order_totals.sql
{{ config(materialized='materialized_view') }}

SELECT
    customer_id,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount
FROM {{ ref('orders') }}
GROUP BY customer_id

設定で WITH 句のオプションを指定できます。

config:
  materialized: materialized_view
  cluster: analytics
  refresh_interval:
    every: '1 hour'
    aligned_to: '00:00:00'
  partition_by: [region, created_date]
  retain_history: '1 hour'

設定	目的
cluster	特定のクラスターで実行します
refresh_interval	継続的な更新ではなく、スケジュールに基づいて更新します
partition_by	フィルタープッシュダウンのための内部ストレージの並べ替えを指定します
retain_history	一定期間、履歴データを保持します

view

標準的な SQL ビューです。結果をメモリに保存する必要のない、軽量な変換に使用します。

{{ config(materialized='view') }}
SELECT * FROM {{ ref('raw_events') }} WHERE status = 'active'

source

外部データソース（Kafka、PostgreSQL、MySQL）を定義します。dbt の外部（Terraform または SQL など）で作成した既存の CONNECTION が必要です。

-- models/staging/kafka_events.sql
{{ config(materialized='source') }}
FROM KAFKA CONNECTION kafka_connection (TOPIC 'events')
FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection

-- models/staging/pg_source.sql
{{ config(materialized='source') }}
FROM POSTGRES CONNECTION pg_connection (PUBLICATION 'mz_source')
FOR ALL TABLES

FOR ALL TABLES を使用するソースは、サブソースを自動的に作成します。下流モデルで参照できるよう、これらを YAML ファイルで dbt ソースとして定義してください。

モデルがソースのサブソースに依存する場合は、-- depends_on: {{ ref('pg_source') }} を使用して実行順序を強制してください。

source_table

ソースからテーブルを作成する新しいマテリアライゼーションで、ソースのバージョン管理をより適切にサポートします。

{{ config(materialized='source_table', strict_mode=True) }}
FROM SOURCE {{ ref('kafka_events') }} (REFERENCE "events_topic")

strict_mode: True を設定すると、オブジェクトがすでに存在する場合、モデルは再作成をスキップします。これを上書きするには force_recreate: True を使用してください。

sink

外部システムにデータをエクスポートします。既存の CONNECTION が必要です。

{{ config(materialized='sink') }}
FROM {{ ref('order_totals') }}
INTO KAFKA CONNECTION kafka_connection (TOPIC 'order-totals-output')
FORMAT JSON
ENVELOPE DEBEZIUM

sink は専用のスキーマ内に配置する必要があります（strict_mode が有効な場合に強制されます）。ブルー／グリーンデプロイ中、sink は新しい上流定義に自動で切り替えられます。

table

Materialize では CREATE MATERIALIZED VIEW に対応します（テーブルの正式なサポートは予定されています）。明確さを優先する場合は、materialized_view を直接使用してください。

seed

標準的な dbt seed が利用できます。テーブルを作成し、CSV データを挿入します。seed 操作用のクラスターを設定してください。

seeds:
  +cluster: seed_cluster

非対応

• incremental：代わりに materialized_view を使用してください。Materialize は増分処理をネイティブに扱います。
• snapshot：サポートされていません。代わりに retain_history を指定した materialized_view を検討してください。

インデックス

ビュー、マテリアライズドビュー、ソース、テーブルにインデックスを追加すると、ポイント検索を高速化できます。

config:
  indexes:
    - columns: [customer_id]
    - columns: [region, created_date]
      cluster: serving_cluster
    - default: true              # index all columns

インデックスはリレーションの作成後に作成されます。cluster を指定しない場合は、モデルのクラスターが使用されます。

Strict Mode

本番環境の分離ルールを適用するには、モデル設定で strict_mode: True を有効にします。

• ソーステーブルは、他のソーステーブルおよびシードとのみ同じスキーマを共有できます
• シンクは、他のシンクとのみ同じスキーマを共有できます
• ソースは、他のソースとのみ同じスキーマを共有できます
• 標準モデル（ビュー、マテリアライズドビュー）は、ソース、シンク、ソーステーブルとスキーマを共有できません

検証は dbt マニフェストを使用してコンパイル時に行われます。

Blue/Green デプロイ

並行環境にビルドし、アトミックに切り替えることで、ダウンタイムなしのデプロイを実現します。

前提条件

dbt_project.yml でデプロイ先を設定します。

vars:
  deployment:
    default:
      clusters:
        - production_cluster
      schemas:
        - public

dbt の実行ロールは、デプロイ対象のスキーマとクラスターを所有している必要があります。シンクは、デプロイ設定に記載されていない専用のスキーマとクラスターに配置する必要があります。

ワークフロー

1. デプロイ環境を初期化します：

dbt run-operation deploy_init

本番環境から設定、権限付与、デフォルト権限を複製して、<schema>_dbt_deploy と <cluster>_dbt_deploy を作成します。本番スキーマとクラスターが存在すること、それらにシンクが含まれていないこと、実行ロールがそれらを所有していることを検証します。デプロイ用オブジェクトがすでに存在する場合にエラーを無視するには、--args '{ignore_existing_objects: True}' を使用します。

マネージドクラスターの場合は、同じサイズ、レプリケーション係数、スケジュールで新しいクラスターを作成します。アンマネージドクラスターの場合は、サイズとアベイラビリティーゾーンを含むレプリカを複製します。

2. green 環境にビルドします：

dbt run --vars 'deploy: True' --exclude config.materialized:source config.materialized:sink

モデルは _dbt_deploy スキーマとクラスターに透過的にルーティングされます。ソースとシンクは必ず除外してください。

String 'deploy:' is not valid YAML が発生した場合（Windows／PowerShell でよく発生します）は、代わりに --vars "{\"deploy\": true}" を使用してください。

3. ハイドレーションを待機します（推奨）：

dbt run-operation deploy_await

単一の統合クエリでクラスターの準備状況をポーリングします（カタログサーバーの負荷を軽減します）。デフォルトでは、15 秒ごとにポーリングし、ラグが 1 秒未満になるまで待機します。--args '{poll_interval: 30, lag_threshold: "5s"}' で設定できます。

準備状況のチェックでは、次の 3 項目を順に評価します。

• レプリカの健全性：過去 24 時間に全レプリカで 3 回以上の OOM キルが発生したクラスターにフラグを付けます
• ハイドレーションの状態：クラスターごとに、ハイドレーション済みオブジェクト数と総オブジェクト数をカウントします
• ラグのしきい値：ウォールクロックのグローバルラグがしきい値未満であることを確認します

4. 本番環境に昇格します：

dbt run-operation deploy_promote

単一のトランザクション内で、アトミックに ALTER SCHEMA ... SWAP WITH と ALTER CLUSTER ... SWAP WITH を実行します。オブジェクト参照が宙に浮くのを避けるため、クラスターより先にスキーマを切り替えます。同時 DDL の競合（SQLSTATE 40001）が発生した場合は、デフォルトで最大 3 回まで自動的に再試行します。

切り替え後、マクロは上流の依存関係が変更されたシンクを検出し、新しい上流定義に切り替えるため ALTER SINK ... SET FROM を実行します。また、デプロイしたスキーマには、デプロイしたユーザー、タイムスタンプ、git commit SHA を含むコメントを付けます。

引数	デフォルト	用途
dry_run	false	実行せずに切り替えコマンドを表示します
wait	false	切り替え前にハイドレーションを待機します
poll_interval	15	準備状況のチェック間隔（秒）。wait が true の場合に使用します
lag_threshold	"1s"	準備状況の最大ラグ。wait が true の場合に使用します
max_retries	3	同時 DDL 競合時の再試行回数

5. クリーンアップします：

dbt run-operation deploy_cleanup

_dbt_deploy スキーマとクラスターを削除します（CASCADE）。

切り替えたクラスターで実行中の SUBSCRIBE コマンドは中断されます。再試行時、クライアントは新しくデプロイされたクラスターに自動的に再接続します。

クラスター管理マクロ

# Create a managed cluster
dbt run-operation create_cluster --args '{cluster_name: analytics, size: "100cc", replication_factor: 1}'

# Drop a cluster
dbt run-operation drop_cluster --args '{cluster: analytics}'

# Set the active cluster for a session
dbt run-operation set_cluster --args '{cluster: analytics}'

Slim デプロイ

ダウンタイムが許容される開発環境の場合：

1. 各デプロイ後に、本番環境の manifest.json を blob ストレージに保存します
2. CI でマニフェストをダウンロードします
3. 変更されたモデルのみを実行します：

   dbt build --select state:modified+ --state ./ --defer
   

ノードの選択

# Run specific models
dbt run --select "model_name"
dbt run --select "tag:nightly"

# Exclude sources and sinks (common in blue/green)
dbt run --exclude config.materialized:source config.materialized:sink

または、selectors.yml を使用します：

selectors:
  - name: exclude_sources_and_sinks
    default: true
    definition:
      union:
        - method: fqn
          value: "*"
        - exclude:
            - 'config.materialized:source'
            - 'config.materialized:sink'

テスト

標準テストは通常どおり動作します。失敗の保存先を設定します：

store_failures: true
store_failures_as: materialized_view  # or 'view' (default: 'table')

ユニットテストには dbt-materialize v1.8.0+ が必要で、上流の依存関係がデータベースに存在している必要があります：

dbt test --select test_type:unit

ソースの鮮度チェックには対応していません（Materialize のソースは定義上、常に最新です）。

モデルコントラクト

コンパイル時に列の型を検証するには、コントラクトの適用を有効にします：

models:
  - name: order_totals
    config:
      contract:
        enforced: true
    columns:
      - name: customer_id
        data_type: text
        constraints:
          - type: not_null    # the only enforced constraint

適用されるのは not_null のみです（ASSERT NOT NULL 経由）。その他すべての制約タイプ（unique、primary_key、foreign_key、check）には対応していません。

注意点

• threads は 1 にする必要があります：Materialize は DDL セッションの並行実行に対応していません。threads を 1 より大きくするとエラーが発生します。
• トランザクションはありません：オートコミットは常に有効です。一部の dbt 内部マクロは異なる動作をする場合があります。
• LIMIT のパフォーマンス：dbt show は LIMIT を使用するため、Materialize のパフォーマンスに影響します。必要な場合にのみ使用してください。
• 接続は外部で管理します：ソースとシンクには、事前に作成された CONNECTION が必要です（Terraform、SQL、その他のツールで作成します）。dbt は接続を管理しません。
• クラスターは事前に存在している必要があります：dbt は暗黙的にクラスターを作成しません。モデルを実行する前に、SQL、Terraform、または create_cluster マクロを使って作成してください。
• Grant には対応していません：dbt の組み込み grants 設定を使用するとエラーが発生します。RBAC には Terraform または SQL を使用してください。
• インクリメンタルには対応していません：代わりに materialized_view を使用してください。Materialize はインクリメンタル処理をネイティブに行います。
• ブルー／グリーンにはシンクの分離が必要です：シンクは、デプロイ設定に記載されていない専用のスキーマとクラスターに配置する必要があります。
• リレーション名の上限：255 文字です（PostgreSQL の 63 文字より大幅に長いです）。