dbt operations の使い方: 管理タスクの自動実行

dbt operations（dbt run-operation）は、モデル実行とは独立して Jinja マクロを直接呼び出すための仕組みです。権限付与、スキーマ作成・削除、メタデータ更新などの管理系タスクを自動化できます。

本稿では、安定機能である run-operation の基本、ベストプラクティス、よくある実装例、スケジューリング方法をまとめ、Analytics Engineer 試験に必要な観点も補足します。

operations の基本と設計指針

operations は dbt の DAG 外でマクロを実行します。実行時には target（プロファイル/接続情報）、env_var、flags、adapter、run_query などのコンテキストが利用可能です。モデル依存関係の解決は行われないため、管理系の一括処理やメンテナンスに向きます。

設計の要点は冪等性と安全な引数設計です。dry-run を備え、明示的なスキーマやロールのみを対象にし、実行ログと戻り値を機械可読な形式（JSON）で返すと、オーケストレーションや監査に強くなります。

• 実行コマンド例: dbt run-operation <macro_name> --args '{"key": "value"}'
• 利用できる代表的コンテキスト: target, env_var, flags, adapter, run_query, log, exceptions
• ユースケース: 権限付与、スキーマ作成/削除、タグ・環境ごとの初期化、外部システムへの通知

operations 実行フロー（概念）

最小テンプレート（戻り値をJSONで返す）

{% macro op_template(required_arg, optional_arg='') %}
  {% if required_arg is not string %}
    {% do exceptions.raise_compiler_error('required_arg must be string') %}
  {% endif %}
  {% do log('target: ' ~ target.name ~ ' (' ~ target.type ~ ')', info=True) %}
  {% do return({'ok': True, 'arg': required_arg, 'opt': optional_arg}) %}
{% endmacro %}

実例1: 権限付与を自動化するマクロ

環境ごとにスキーマへの USAGE、テーブル/ビューへの SELECT を付与する運用は典型的な operations の適用先です。Snowflake と Databricks（Unity Catalog 含む）向けの単純化した例を示します。実行前に dry_run で発行SQLを確認できるようにします。

BigQuery では IAM ロール付与が主となるため、同等の処理は GRANT 文ではなくプロジェクト/データセット レベルの権限管理で行うのが一般的です。以下のマクロは Snowflake と Databricks/Spark に限定しています。

• 引数: target_schema, role, dry_run（既定: false）
• 対象: Snowflake, Databricks/Spark。その他のアダプタではエラーにする
• 出力: 実行有無とステートメント数（または一覧）

grant_usage_and_select マクロ例

{% macro grant_usage_and_select(target_schema, role, dry_run=False) %}
  {%- set wh = target.type -%}
  {%- set statements = [] -%}
  {%- if wh == 'snowflake' -%}
    {%- do statements.append('grant usage on schema ' ~ target.database ~ '.' ~ target_schema ~ ' to role ' ~ role) -%}
    {%- do statements.append('grant select on all tables in schema ' ~ target.database ~ '.' ~ target_schema ~ ' to role ' ~ role) -%}
    {%- do statements.append('grant select on future tables in schema ' ~ target.database ~ '.' ~ target_schema ~ ' to role ' ~ role) -%}
    {%- do statements.append('grant select on all views in schema ' ~ target.database ~ '.' ~ target_schema ~ ' to role ' ~ role) -%}
    {%- do statements.append('grant select on future views in schema ' ~ target.database ~ '.' ~ target_schema ~ ' to role ' ~ role) -%}
  {%- elif wh in ['databricks','spark'] -%}
    {%- set cat = target.get('catalog') if target.get('catalog') else target.database -%}
    {%- set full_schema = cat ~ '.' ~ target_schema -%}
    {%- do statements.append('grant usage on schema ' ~ full_schema ~ ' to ' ~ role) -%}
    {%- do statements.append('grant select on all tables in schema ' ~ full_schema ~ ' to ' ~ role) -%}
    {%- do statements.append('grant select on all views in schema ' ~ full_schema ~ ' to ' ~ role) -%}
  {%- else -%}
    {%- do exceptions.raise_compiler_error('Unsupported adapter for this macro: ' ~ wh) -%}
  {%- endif -%}

  {%- if dry_run -%}
    {%- for s in statements %}{% do log(s, info=True) %}{% endfor -%}
    {% do return({'executed': False, 'statements': statements}) %}
  {%- else -%}
    {%- for s in statements %}
      {% do run_query(s) %}
      {% do log('OK: ' ~ s, info=True) %}
    {%- endfor -%}
    {% do return({'executed': True, 'count': statements|length}) %}
  {%- endif -%}
{% endmacro %}

-- 実行例（CLI）
-- 事前に profiles.yml の prod ターゲットを設定
-- dry-run で確認
-- dbt run-operation grant_usage_and_select --target prod --args '{"target_schema": "analytics", "role": "ANALYST_ROLE", "dry_run": true}'
-- 本実行
-- dbt run-operation grant_usage_and_select --target prod --args '{"target_schema": "analytics", "role": "ANALYST_ROLE"}'

実例2: スキーマのライフサイクル自動化（作成/削除の安全運用）

検証環境の一時スキーマ作成、検収後のクリーンアップなど、スキーマのライフサイクル操作も operations で自動化できます。blast radius を最小化するため、明示リストのみに作用し、dry-run を通すポリシーにします。

以下はスキーマの create/drop をまとめて行うマクロ例です。多くの DWH で create schema if not exists / drop schema if exists が利用できます。

• 引数: schemas（文字列または配列）、action（create/drop）、dry_run
• 明示されたスキーマのみ対象。ワイルドカードや推測は避ける
• 削除時は cascade の有無を DWH ポリシーに合わせて調整

ensure_schemas マクロ例

{% macro ensure_schemas(schemas=[], action='create', dry_run=False) %}
  {%- if schemas is string -%}
    {%- set schemas = [schemas] -%}
  {%- endif -%}
  {%- set stmts = [] -%}
  {%- for s in schemas -%}
    {%- if action == 'create' -%}
      {%- set stmt = 'create schema if not exists ' ~ target.database ~ '.' ~ s -%}
    {%- elif action == 'drop' -%}
      {%- set stmt = 'drop schema if exists ' ~ target.database ~ '.' ~ s ~ ' cascade' -%}
    {%- else -%}
      {%- do exceptions.raise_compiler_error("action must be 'create' or 'drop'") -%}
    {%- endif -%}
    {%- do stmts.append(stmt) -%}
  {%- endfor -%}

  {%- if dry_run -%}
    {%- for s in stmts %}{% do log(s, info=True) %}{% endfor -%}
    {% do return({'executed': False, 'statements': stmts}) %}
  {%- else -%}
    {%- for s in stmts %}{% do run_query(s) %}{% endfor -%}
    {% do return({'executed': True, 'count': stmts|length}) %}
  {%- endif -%}
{% endmacro %}

-- 実行例
-- dbt run-operation ensure_schemas --target dev --args '{"schemas": ["tmp_feature_a", "tmp_feature_b"], "action": "create", "dry_run": true}'

自動実行: dbt Cloud と外部オーケストレーション

dbt Cloud の Job では、Step に run-operation を追加できます。例: Step1 で権限付与、Step2 で本番パイプライン（dbt build）、Step3 で後片付けといった並びが定番です。スケジュールやトリガにより安定的に実行され、ログ/結果は Cloud 上で確認できます。

CI/CD（GitHub Actions など）や外部オーケストレータ（Airflow, Dagster など）からも、同じ CLI を呼び出せば再現性を保てます。標準出力の JSON を収集し、監査ログに保存しておくと後から検証しやすくなります。

• dbt Cloud: Job の Step に dbt run-operation ... を追加
• GitHub Actions: ランナーで dbt とアダプタをインストールし、CLI 実行
• Airflow: BashOperator または KubernetesPodOperator で CLI 実行、XCom で結果を保存

GitHub Actions 例（Snowflake アダプタ）

name: run-operations
on:
  workflow_dispatch:
jobs:
  op:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dbt core and adapter
        run: |
          pip install --upgrade pip
          pip install dbt-core dbt-snowflake
      - name: Run dbt operation (dry-run)
        env:
          DBT_USER: ${{ secrets.DBT_USER }}
          DBT_PASSWORD: ${{ secrets.DBT_PASSWORD }}
        run: |
          dbt deps
          dbt debug --target prod --profiles-dir .
          dbt run-operation grant_usage_and_select \
            --target prod \
            --args '{"target_schema": "analytics", "role": "ANALYST_ROLE", "dry_run": true}'

安全性とベストプラクティス

運用系マクロは失敗時の影響が大きくなりがちです。dry-run の実装、対象の明示、実行前ログ、失敗時の早期中断（compiler_error）を基本に据えます。必要に応じてトランザクションでまとめるか、逆に DDL の性質に合わせて小刻みに実行し失敗箇所を特定できるようにします。

入力検証は厳密に行い、環境変数や target による分岐は最小限にします。権限付与は最小権限の原則を守り、将来オブジェクト（future grants など）を適切に使います。

• 冪等性: 同じ操作を繰り返しても整合状態が変わらないようにする
• dry-run: すべての SQL をログに出し、レビュー可能にする
• 入力検証: スキーマ名/ロール名のスキーマチェック、空値拒否
• blast radius 最小化: ワイルドカード削除を禁止、明示リストのみ対象
• 失敗時の中断: exceptions.raise_compiler_error で安全にフェイル
• 監査: return() の JSON を保存して追跡可能性を確保

トランザクション適用の簡易ラッパ（対応するDWHのみ）

{% macro run_in_transaction(sqls=[]) %}
  {# 注意: 一部DWHはDDLのトランザクションをサポートしません #}
  {% do run_query('begin') %}
  {% for s in sqls %}
    {% do run_query(s) %}
  {% endfor %}
  {% do run_query('commit') %}
  {% do return({'executed': sqls|length}) %}
{% endmacro %}

-- 利用例
-- {% do run_in_transaction(['grant usage on schema DB.SCHEMA to role R', 'grant select on all tables in schema DB.SCHEMA to role R']) %}

ロギングと戻り値の扱い

dbt run-operation で実行されたマクロは、return() に渡したオブジェクトを標準出力に表示します。これを機械可読な JSON にしておくと、外部からの収集・検証が容易です。合わせて log() で人間が読みやすいメッセージも出力します。

CLI 実行結果から JSON 部分のみを抽出し、保存・通知に使う例を示します。環境によっては --quiet オプションを活用し、ノイズの少ない出力にすると扱いやすくなります。

• return() は 1 回だけ呼び出すのが実務上扱いやすい
• ログは info レベルで重要SQL、debug レベルで詳細を分離
• 外部からは jq 等で JSON を取り出し、監査用に保存

標準出力から JSON を抽出する例（bash）

# dry-run の結果を取得
out=$(dbt --quiet run-operation grant_usage_and_select --args '{"target_schema": "analytics", "role": "ANALYST_ROLE", "dry_run": true}')
# dbt は "The result of this operation was: <JSON>" の形式で出力するため、JSON 部分を抽出
json=$(printf "%s\n" "$out" | sed -n 's/.*The result of this operation was:\s*//p')
# 任意: 整形や保存
printf "%s\n" "$json" | jq .
echo "$json" > operation-result.json

問題で確認

よくある質問