dbt CloudのジョブをAPIを介して実行してみた
かわばたです。
dbt CloudのジョブをAPIを介して実行する方法を試していきます。
最近UIも変わったこともあり、改めてどのような形で実行できるか確認していきます。
【公式ドキュメント】
対象読者
- dbt CloudのジョブをAPIを介して実行する方法に興味のある方
検証環境と事前準備
検証環境
- dbt CloudのEnterprise版
- SnowflakeトライアルアカウントのEnterprise版
- Fivetranアカウント
事前準備
dbtのサンプルデータであるjaffle_shopを活用しています。
概要
dbt Administrative APIは、Enterpriseプランでデフォルトで有効化されていますが、StarterプランでもAPIアクセスは可能です。
dbt Administrative APIにはバージョンv2およびv3があります。基本的にはv3推奨ですが、v2にしかない機能も存在するため確認が必要です。
APIによる自動化は、以下のようなユースケースがあります。
- イベント駆動型パイプライン: FivetranやAirbyteなどの上流のデータ取り込みプロセスが正常に完了したことをトリガーとして、dbtジョブを自動的に実行します。
【参考記事】
- 高度なオーケストレーション: Apache Airflow、Prefect、Dagsterといった外部のオーケストレーションツールが管理する複雑なワークフローにdbt Cloudをシームレスに統合することができます。
【参考記事】※こちらの例はdbt Core
APIトークンについて
dbt Cloud APIへの認証にはトークンが使用され、2つの主要なタイプが存在します。
- パーソナルアクセストークン：特定のユーザーアカウントに紐づき、そのユーザーの権限を継承。
- サービスアカウントトークン：dbt Cloudアカウント自体に紐づき、ユーザーから独立した特定の権限セットが割り当て。
基本的には、サービスアカウントトークンの利用を推奨しています。
理由としては、ユーザーに紐づいていないため退職や異動などが生じてもパイプラインを維持することが可能です。
| 特徴 | パーソナルアクセストークン (PAT) | サービスアカウントトークン |
|---|---|---|
| スコープ | ユーザーレベル（特定のユーザーアカウントに紐づく） | アカウントレベル（ユーザーから独立） |
| 権限 | 作成したユーザーの全権限を継承 | 特定の権限セット（例：Job Admin）が割り当てられる |
| ライフスパン | ユーザーが無効化されるとトークンも無効になる | 明示的に失効させない限り有効 |
| ユースケース | 開発、CLI、APIテスト | 本番自動化、CI/CD |
実際に試してみた
サービスアカウントトークンの生成
-
dbt CloudのUIで、画面左上のメニューから
Account Settingsを選択します。
Service Tokensをクリックし、画面右上の+ Create service tokenをクリックします。
-
+ Add permissionをクリックし、Permission Setを選択します。ここでは、Job Adminを選択していますが下記ドキュメントを確認いただき、最小権限の原則に則ることを推奨します。
また、ProjectもAll Projectsではなく任意のプロジェクトを選択しています。
【Permission Sets ドキュメント】
- 内容に問題がなければ、
Saveをクリックしトークンを作成します。
アカウントIDとジョブID
アカウントID
- dbt CloudのUIで、画面左上のメニューから
Account Settingsを選択します。
Accountをクリックし、Account informationのAccount IDを確認します。
もしくは次のジョブIDと一緒に確認しましょう。
ジョブID
- dbt CloudのUIで、画面左上のメニューから
Orchestration→Jobsを選択し、任意のジョブを選択します。
API triggerをクリックすると、Account IDとJob IDを確認できます。
cURLで実行
ターミナルで実行していきます。
以下のコマンドを実行する際に、YOUR_ACCOUNT_ID,YOUR_JOB_ID,YOUR_DBT_CLOUD_TOKENを自身の値に置き換えてください。
curl --request POST \
--url https://cloud.getdbt.com/api/v2/accounts/YOUR_ACCOUNT_ID/jobs/YOUR_JOB_ID/run/ \
--header 'Authorization: Token YOUR_DBT_CLOUD_TOKEN' \
--header 'Content-Type: application/json' \
--data '{"cause": "Triggered via cURL for testing"}'
dbt Cloudを確認すると、先ほど作成したサービスアカウントトークンがトリガーとなっていることが確認できます。
Fivetranとの連携
APIを利用して各種製品との連携が可能ですが、ここではFivetranを活用して、Fivetranの同期が完了したタイミングでdbt Cloudのジョブを実行するような仕組みを試していきます。
-
Fivetranの
Transformationsをクリックし、右上側のAdd Transformationをクリックします。
-
任意の
destinationを選択します。
Orchestrate your connected projectsの項目でdbt CloudのConnect projectを選択します。
Region,Service Token,Account,Projectを選択します。
- 以下のような画面が出ましたら、接続が完了しています。
- 実行するjobを選択し、
Select Schedule typeをintegratedを選択、Select trigger connectionsでコネクションを選択します。
integratedを選択することで、コネクションが同期が完了したタイミングでdbt Cloudのジョブを実行することができます。
※今回は検証ということで、関係のないGoogle Sheetsを使用しています。
- コネクションを以下のように同期しました。
コネクションが完了したタイミングで、実行されていました。
Snowflake側も確認し、ジョブが正常に実行できていることを確認しました。
最後に
dbt Cloud APIを利用してジョブの実行を検証していきました。
私の中で重要な観点を以下にまとめます。
- セキュリティ：個々のユーザーから独立したサービスアカウントトークンを使用することを推奨します。
- APIは非同期：ジョブのトリガーは「発行して終わり」の非同期プロセスであるため、実行が成功したか否かを知るためのプロセスは必要です。
この記事が何かの参考になれば幸いです！
