dbt core + Github Actions で PR ごとに独立したテスト環境を自動構築する

はじめに

AI Native な開発の浸透とともに、エンジニアの手元には実装そのものに費やす時間以外の余白が増えてきました。OLTA でもその時間を活かして、本来の担当領域からデータ領域へ染み出していくエンジニアが増えています。

OLTA のデータ基盤は dbt core で構築しており、複数プロダクトにまたがる多数のモデルが上流から下流へ依存し合いながら日次バッチでビルドされています。データ基盤への変更頻度が上がるのは喜ばしい一方で、副作用として「変更のたびに動作確認の安定性が揺らぐ」という新しい困りごとが見えてきました。具体的には次のような状況です。

動作確認は開発者個人の手元での dbt build に依存している
既存の CI は静的解析・構文解析までしか行っていないため、モデル実行時にしか現れないエラーに気づくのは merge 後 になる

実際、ある PR の変更が下流モデルへ意図せず波及し、翌日の本番バッチが落ちる事故が起きました。

本記事では、これらの課題を解決するために Slim CI を実装した話を紹介します。

Slim CI とは

Slim CI は、PR ごとに変更したモデルとその下流だけをビルドする手法です。未変更の上流モデルは本番テーブルをそのまま参照することで、CI 全体の実行時間と BigQuery スキャンコストを抑えます。dbt core では --select state:modified+ --defer --state <path> というコマンドで起動することで実現します。

state:modified+ … 前回のビルド状態 (本番の manifest.json) と比較し、変更されたモデルとその下流すべて を選びます
--defer --state <path> … 変更対象に含まれない上流モデルは、本番の物理テーブルをそのまま参照 します

ビルド先のデータセットをどう切るか・後始末をどうするかは Slim CI 自体の定義には含まれず、運用要件として個別に設計する必要があります。今回はそこも含めて「PR ごとに独立した一時データセットを発行し、PR が閉じたら自動で削除する」という形で組み込みました。

今回直面した課題

筆者たちが解きたかった課題は次の 3 つに整理できます。

PR の段階で、変更したモデルとその下流が本番と同じスキーマ・データに対して実行可能であることを確認したい: 既存 CI は SQL のフォーマットや dbt の構文解析だけを実行しており、上流の変更が下流に波及してカラムが重複するような 実行時にしか現れない整合性エラー は本番反映後に初めて顕在化していた
本番データセット全体を再ビルドするのは現実的ではない: 本番のフルビルドは時間とスキャンコストが大きく、PR 毎に走らせたくない
検証用のテーブル・ビューが本番や開発者個人のデータセットを汚さない: PR ごとに独立した名前空間でリソースを作成・削除することで、本番環境に影響を与えない仕組みが欲しい

Slim CI でどう解決するか

これら 3 つに対して、Slim CI と今回の運用設計はそれぞれ次のように応えます。

state:modified+ で変更モデルとその下流を実 SQL で BigQuery にビルドするため、上流の変更が下流に波及して起きるカラム重複や型不整合などの 実行時にしか現れない整合性エラー も CI の段階で検知できます
--defer --state で未変更の上流は本番を参照するため、BigQuery のスキャン量は変更分のみに抑えられます
PR 番号などをキーに PR ごとに独立した一時データセット を発行し、PR が閉じたタイミングで自動削除することで、本番や個人データセットを汚しません

dbt Cloud では Slim CI が組み込みで提供されていますが、dbt core で実現する方法をこれから紹介します。本記事のスコープは、本番 manifest の保管、PR ごとの一時データセットの作成と削除、認証、権限境界といった周辺インフラを一通り整える話です。

アーキテクチャの全体像

完成形は次のような構成です。

ポイントは次の 4 つです。

本番環境の定期実行の最後に manifest.json を GCS に保存 する
PR ごとに PR 番号で名前を切った一時データセット を作る
PR が close されたら 一時データセットを削除 する
何らかの理由で削除し損ねた場合の 保険として週次で日数経過分を一掃 する

このワークフローを実際に走らせると、PR を起点に次のような順序で進みます。

PR 作成 / 更新 … Slim CI の workflow が trigger され、変更モデルとその下流が <prefix>_pr_<n> 配下に CREATE される。変更されていない上流モデルは defer で本番データセットを参照する
CI 結果が PR に表示 … 実行時にしか現れない整合性エラーがあれば、ここで CREATE が失敗して検知できる
PR close / merge … Cleanup の Workflow が trigger され、マクロを実行することで対応する一時データセットを DROP SCHEMA CASCADE で削除
週次の定期実行 … 14 日経過した取りこぼしを一括で掃除

一時データセットの命名と削除

PR ごとに一時データセットを作る場合、最初に決めるのは命名規則です。今回は PR 番号をプレフィックスに含めた <prefix>_pr_<PR番号> という形を採用しました。

profile の ci ターゲットでは、PR 番号を環境変数で受け取って dataset 名を組み立てます。

ci:
  dataset: "<prefix>_pr_{{ env_var('DBT_CI_PR_NUMBER') }}"
  location: asia-northeast1
  method: oauth
  priority: interactive
  project: <gcp-project-id>
  threads: 4
  type: bigquery

次に削除の設計ですが、BigQuery には default_table_expiration_ms という設定があり、これを使えばデータセット内のテーブルは時間経過で自動的に消えます。しかしデータセットそのものの自動 expiration は提供されていません。つまり何もしないと、空のデータセットだけが永遠に残り続けます。

そのため、cleanup は明示的に DROP SCHEMA CASCADE を発行する設計にして、dbt のマクロとして次のように書きます。

{% macro drop_ci_dataset(pr_number, dry_run=True) %}
    {% if pr_number is none or pr_number == "" %}
        {{ exceptions.raise_compiler_error("pr_number is required") }}
    {% endif %}

    {% set prefix = "<prefix>_pr_" ~ pr_number %}
    {% set project_id = target.project %}
    {% set location = target.location %}

    {% set find_datasets_sql %}
    select schema_name
    from `{{ project_id }}`.`region-{{ location }}`.INFORMATION_SCHEMA.SCHEMATA
    where schema_name = '{{ prefix }}'
       or starts_with(schema_name, '{{ prefix }}_')
    {% endset %}

    {# dbt parse 時の None 参照を避けるため execute でガード #}
    {% if execute %}
        {% set datasets_to_drop = run_query(find_datasets_sql) %}
    {% else %}
        {% set datasets_to_drop = [] %}
    {% endif %}

    {% if not dry_run %}
        {% for row in datasets_to_drop.rows %}
            {% set drop_sql = "drop schema if exists `"
                ~ project_id ~ "`.`" ~ row["schema_name"] ~ "` cascade" %}
            {% call statement("drop_" ~ loop.index) %}
                {{ drop_sql }}
            {% endcall %}
        {% endfor %}
    {% endif %}
{% endmacro %}

このマクロを、PR close を契機に GitHub Actions の workflow から dbt run-operation で呼びます。

name: dbt Slim CI Cleanup
on:
  pull_request:
    types: [closed]
    branches: [main]

jobs:
  cleanup:
    runs-on: ubuntu-latest
    if: github.event.pull_request.head.repo.fork == false
    steps:
      # ... auth / setup ...
      - working-directory: <dbt-project-dir>
        env:
          DBT_PROFILES_DIR: ./profiles
          DBT_CI_PR_NUMBER: ${{ github.event.pull_request.number }}
        run: |
          uv run dbt run-operation drop_ci_dataset \
            --args "{pr_number: ${{ github.event.pull_request.number }}, dry_run: false}" \
            --target prod

pull_request types: [closed] にして merge と close の両方で発火させるようにします。

ただし、何らかの理由で GitHub Actions の cleanup が動かないこともあります。そうしたケースの保険として「14 日以上経過した一時データセットを一括で削除する」マクロを別途用意し、定期実行ジョブから呼んでいます。

IAM condition で「触れる名前空間」を構造的に縛る

Github Actions でクリーンアップを成立させるには、CI 用のサービスアカウントに bigquery.datasets.delete 権限が必要です。しかしこれをそのまま付与すると、認証情報が漏れた場合に本番データセットを丸ごと削除されかねません。

そこで、bigquery.datasets.delete だけを持つ専用カスタムロール を用意し、そのバインディングに IAM condition を付けて削除対象のプレフィックスを縛る 方法を採りました。

resource "google_project_iam_custom_role" "dataset_delete_role" {
  role_id     = "datasetDeleteRole"
  title       = "BigQuery dataset delete (condition で対象を限定)"
  permissions = ["bigquery.datasets.delete"]
}

resource "google_project_iam_member" "ci_sa_conditional_deleter" {
  project = var.project_id
  role    = google_project_iam_custom_role.dataset_delete_role.id
  member  = "serviceAccount:<ci-sa>"

  condition {
    title      = "CI 用プレフィックスを持つデータセットのみ削除可能"
    expression = "resource.name.startsWith(\"projects/${var.project_id}/datasets/<prefix>_pr_\")"
  }
}

これで、たとえ CI 用 SA の鍵が漏洩した場合でも、削除できるのは <prefix>_pr_* のデータセットだけ に構造的に縛られます。本番や開発者個人のデータセットに触ることはできません。

IAM 設計をするときに「ロール = 何ができるか」「condition = どこに対してできるか」を切り分けて考えるのは GCP の基本作法ですが、これを徹底すると CI の権限境界がきれいに引けるので強くおすすめできます。設定後は GCP Console の IAM 一覧画面では condition の存在が目立たないので、CLI で網羅的に検証することをおすすめします。

gcloud projects get-iam-policy <project> --format=json \
  | jq -r '.bindings[]
     | select(.members[]? | contains("<ci-sa>"))
     | {role, condition: (.condition.title // "no condition")}'

効果

導入後、データ基盤を触る開発者の体験は次のように変わりました。

PR の段階で実行時にしか現れない整合性エラーが見える: 変更モデルの下流まで含めて実際に SQL が実行されるので、上流の変更が下流に波及して起きるカラム重複や型不整合は CI によって気づけるようになりました
検証用環境を手動で用意する手間がなくなった: 以前は変更を試したい時、個人データセットを切り替えて手元でビルドし直す、というフローでした。Slim CI が PR ごとに独立した名前空間へ自動でビルドしてくれるので、開発者もレビュアーもそのデータセットに対して直接 bq query を投げて挙動を確認できます。レビューが「コードを読むレビュー」から「動くものを触りながら確認するレビュー」に変わったのは想定外の副産物でした
コストが小さい: defer で未変更上流は本番を読むため、BigQuery のスキャン量は変更モデル分のみに抑えられます。本番のフルビルドは 10 分程度かかる規模感ですが、典型的な PR の Slim CI は 1 分前後で完結します
データ領域に染み出すエンジニアの心理的安心感が上がった: 本番反映前に「実体に対する CI 結果」と「PR ごとの検証データセット」の両方が手に入るため、データ基盤に普段触らないエンジニアでも「壊さないか」という不安を抱えずに PR を出せるようになりました。これは冒頭で書いた本来の困りごとに対する直接の応答です

今後について

現状はあえて Slim CI を「マージ必須チェック」(GitHub の required status check) には登録していません。CI が必ず走って結果が PR に表示されること自体は確保しつつ、マージをブロックするかどうかは開発者の判断に委ねる、というレベル感です。マージ必須化は試運転を続けながら、すり抜けの実例が見えてきたタイミングで検討しようと考えています。

中長期では、もう少し踏み込んだ活用も視野に入れています。

PR レビューでの動作確認の標準化: 現状は副産物として「PR ごとの一時データセットを使った動作確認」が成立していますが、レビューフローとしてはまだ明文化されていません。「PR を見たら対象データセットを覗く」をレビュー手順の一部にしていくと、コードだけでは追いきれない実データに対する妥当性チェックを習慣化できます
データ品質テストの本格組み込み: 本記事のスコープでは扱いませんでしたが、dbt test (relationships / not_null / unique など) を Slim CI に組み込めば、ロジックの妥当性まで PR の段階で見えるようになります。スキャンコストとのバランスを取りながら段階的に拡張していくのが次の宿題です
他リポジトリへの横展開: 今回の構成は OLTA の dbt リポジトリ専用に組みましたが、Slim CI の実行・cleanup・IAM の設計はそのまま別の dbt プロジェクトにも転用できます。データ基盤の品質を組織横断で底上げするレシピとして整理し直す価値があります

まとめ

dbt core で Slim CI を組むのに必要なのは、突き詰めると以下の 4 点でした。

本番状態の manifest.json をオブジェクトストレージに保管する仕組み
PR ごとに名前空間を切った一時データセットの作成と削除
それを実行する CI workflow と認証経路 (Workload Identity Federation)
強権限をうっかり付与しないための IAM condition による削除対象プレフィックスの構造的制限

dbt の機能としては state:modified+ --defer --state を中心に揃っているので、周辺インフラの配管がそろえば、dbt Cloud を使っていなくても PR ごとの独立検証環境を運用に乗せられます。「全モデルビルドのコスト」「PR ごとの独立テスト環境」というよくある課題に対する、再現性のあるレシピとして参考になれば幸いです。

OLTA TECH BLOG

テクノロジーと好奇心で事業を成長させる