はじめに
前回の記事ではdbtのドキュメントを参考にGitHub Actionsを用いてCI/CDを実装しました。
上記の記事で作成したワークフローでは、dbtのjobが実行されるのはpull requestがマージされた時のみです。
しかし開発中に「現在のブランチでdbt jobの動作確認をしたい」という場面があると思います。
dbt Cloud上でブランチを指定してjobを実行する方法は、私が調べた中では見つけることはできませんでした。（もし知ってる方がいましたらご教授ください。）
dbt Cloud APIを利用すればブランチを指定できるので、この記事ではブランチを指定してdbt jobを実行するワークフローをGitHub Actionsで実装していきます。
pushの度にdbt jobを実行するのは頻度が高くなりすぎるので、今回は好きなタイミングで手動実行できるようにしようと思います。
概要
ワークフローの中身は前回の記事とほとんど同じ構成で、手動実行するように変更したワークフローを実装していきます。
実装
.github/workflows/配下にrun_job_manually.ymlというファイルを作成し、以下を記述します。
name: run dbt Cloud job manually
# dbt jobを手動で実行
# ブランチとjob IDを指定する必要あり
on:
workflow_dispatch:
inputs:
job_id:
type: string
description: 'Job ID'
required: true
jobs:
# jobを実行するためにdbt Cloud APIを呼び出す
run_dbt_cloud_job:
name: Run dbt Cloud Job
runs-on: ubuntu-latest
# 実行に必要な環境変数を設定
env:
DBT_ACCOUNT_ID: 00000 # enter your account id
DBT_PROJECT_ID: 00000 # enter your project id
DBT_PR_JOB_ID: ${{ inputs.job_id }}
DBT_API_KEY: ${{ secrets.DBT_API_KEY }}
DBT_JOB_CAUSE: 'GitHub Pipeline CI Job Manually'
DBT_JOB_BRANCH: ${{ github.ref_name }}
steps:
- uses: "actions/checkout@v3"
- uses: "actions/setup-python@v2"
with:
python-version: "3.9"
- name: Run dbt Cloud job
run: "python python/run_and_monitor_dbt_job.py"
前回の記事で作成したワークフローとの違いは以下です。
- ワークフローを手動実行するためにトリガーイベントをworkflow_dispatchにして、入力としてjob_idを求めている点
- 環境変数の
DBT_PR_JOB_IDに対して、ワークフロー実行時に入力されたジョブIDを渡している点
アカウントIDとプロジェクトIDを入力してマージまで行い、この変更をmainブランチに反映させておきます。
動作確認
実装するワークフローとしては以上のみですが、ここでは正しくブランチとdbtのjob IDを渡してテストできるかを確認していきます。
利用するデータ
私の環境では接続先としてSnowflakeを用いていて、ソースとなるLINEARテーブルには以下のデータが入っています。
-
xカラム: 0~99の100個の整数 -
yカラム: xを2倍して、それに0~49の中からランダムに選んだ整数を足した値
イメージとしてはy = 2xの線形関数にノイズが加わったものです。
データの生成処理にはSnowflakeのpython worksheetsを利用しました。
具体的なコードは以下の記事で紹介しています。
実施するテスト
今回はyカラムに対して、以下の2つのテストを行います。
-
not_null: nullがないこと -
not_negative: 負数がないこと
not_nullはdbtのデフォルトのテストで、not_negativeの方は自分でカスタムテストを作成していきます。
not_nullではテストが成功して、not_negativeではテストが失敗するようにしたいので、あらかじめデータの1つを負数に置き換えておきます。
update SOURCE_DB.PUBLIC.LINEAR set "y" = -10 where "x" = 50;
x=50のデータだけyが負数になっているのが分かります。
dbt実装
次にdbtでモデル等を実装します。
sourceとしてSnowflakeのテーブルを指定し、
version: 2
sources:
- name: source_db_public
database: source_db
schema: public
tables:
- name: linear
Snowflakeから単純にデータを取ってくるだけのモデルを作成します。
select
"x" as x,
"y" as y
from
{{ source("source_db_public", "linear") }}
まずはyカラムにnot_nullテストのみを設定します。
version: 2
models:
- name: linear_model
description: "線型データ"
columns:
- name: y
description: "欠損値チェック"
tests:
- not_null:
config:
store_failures: true
ここで一度mainブランチに反映させておきます。
動作確認のため、この時点で手動でワークフローを実行しdbt jobをテストしてみるとnot_nullテストのみが実施されますが、yカラムに欠損値はないためテストは成功します。
次にnot_negativeテストを加えていきます。
tests/generic/配下にnot_negative.sqlを追加し、対象モデルから負数のデータを取得する（= 負数があると失敗する）テストを追加します。
{% test not_negative(model, column_name) %}
select * from {{ model }} where {{ column_name }} < 0
{% endtest %}
そのテストも実施対象に含めるためにpropertiesファイルに追記します。
version: 2
models:
- name: linear_model
description: "線型データ"
columns:
- name: y
description: "欠損値・負数チェック" # 負数のチェックも含めることを追加
tests:
- not_null:
config:
store_failures: true
# -- 以下を追加 --
- not_negative:
config:
store_failures: true
この状態でコミット&pushします。
pull requestを作成する前にここでGitHubからワークフローを実行し、今回のコミットで変更された内容についてdbt jobが成功するかを確認しておきます。
リポジトリのActionsページ上で、作業していたブランチを選択しJob IDにはテストしたいジョブIDを指定してワークフローを実行します。
私がここで指定しているjobは、特定のモデル（今回だとlinear_model）に対してdbt runとdbt testを実行するだけの単純なものです。
ワークフローを実行するとエラーになりました。
ログを参考にdbt jobの履歴を見てみると、今回のコミットで作成したnot_negativeテストで失敗していることが分かります。
今回はstore_failures: trueを指定しテストで検知されたデータをテーブルに格納しているので、そのテーブルを見てみるとx=50のデータが検知されていることが分かります。
このようにワークフローやdbt jobがエラーになることで、実装したテストか使用したデータに問題があることが分かります。
今回はデータにわざと間違った値を入れているのでデータに問題があることが原因ですが、実装したテストに問題がある場合はpull requestを作る前にコード修正の必要性に気づけます。
まとめ
この記事では手動でdbt jobを実行するワークフローを作成しました。
今回作成したコードは以下で参照できます。
pull request作成前にjobを動かせるのはエビデンス作りにも役立つのかなと思います。
私自身まだdbt歴が浅く、より良い運用方法を模索中なので、「こんな運用方法あるよ！」という方はぜひご意見ください。