テックタッチアドベントカレンダー 13 日目を担当しますデータエンジニアの acchan です。 この間完全個室型のサウナに初めて行ってきまして、贅の極みを堪能してきました。みなさんもこの 1 年間頑張ってきた自分のご褒美にいかがでしょうか。
はじめに
弊社では最近、データ分析基盤のモデリング用途として dbt を使い始めました。まだ間もないですが、モデルの共同開発が以前より活発になり、導入したことによるメリットが実感として現れつつあります。 しかし、作成されたモデルは本当に正しいデータを持っているのか？定義された指標通りに集計がされているのか？といった品質チェックやモデルそのものの最新性などの監視といった運用上の課題が残っています。
この問題を解消できないか調査しているうちに Elementary というツールに行き着きましたので、今回は Elementary 公式の Tutorial を通して dbt で生成したモデル内の異常値検知（+ Slack へのアラート通知）を試してみます。
Elementaryとは
Elementary はオープンソースの dbt 向けのデータ監視・モニタリングツールです。 データやモデルの異常検出・アラート送信が行えるほか、モデルパフォーマンスの可視化やリネージュを通してモデルの持つ問題の根本原因をすばやく把握する事ができます。
Elementary は dbt パッケージと Elementary CLI の 2 つのコンポーネントからなります。前者ではdbt testを通したデータ異常値検出の結果やdbt runの実行結果などを DWH 上の Elemantary 用スキーマ内に収集する役割を持ち、後者では主にレポートやアラート送信に使用されます。
チュートリアルをやってみる
下準備
さっそく Tutorial を試してみます。 今回は dbt 公式の公開している Docker イメージをもとにお試し環境を作ってみました。これは dbt と Elementary CLI がすでにインストール済みの環境になります。
サンプルリポジトリを Github に公開しています。
なお、モデルの作成先 DWH として Redshift を想定しています。
ターミナルを起動し、リポジトリのディレクトリ配下にて docker コンテナを起動します。
docker compose up -d
以後、コンテナ内のシェルでコマンドを実行するものとします。
docker compose exec -it dbt_container bash
コンテナ起動後、/usr/app/dbt 配下にdbt プロジェクトを新規作成します。
dbt init elementary_tutorial
Tutorial のサンプル dbt プロジェクトをダウンロードし、先程のプロジェクト（elementary_tutorial）配下にコピーします。
サンプルデータセットを投入します。
cd elementary_tutorial dbt seed
これで実行環境とサンプルプロジェクトの構築は完了です。
パッケージのインストールと Elementary の初期設定
Elementary パッケージのセットアップを行います。プロジェクト配下の dbt_packages.yml に次のブロックを追記します。
packages: - package: elementary-data/elementary version: 0.6.4 ## Docs: https://docs.elementary-data.com
追加したプロジェクトをインポートします。
dbt deps
次に、Elementary で収集する各種データ投入用のモデルを作成します。
dbt_project.yml の models に Elementary 用のスキーマを定義します。
以下の例では <ターゲットスキーマ名>_elementary の名前でスキーマが作成され、その配下にモデルが追加されます。今回は advent_calendar をターゲットスキーマ名として設定しているため、 advent_calendar_elementary という名前になります。
models: ## elementary models will be created in the schema '<your_schema>_elementary' ## see docs: https://docs.elementary-data.com/ elementary: +schema: "elementary"
実際に DWH 上にモデルを追加します。こちらはスキーマ作成の権限を持ったユーザーで実行してください。
dbt run --select elementary
実行後は次のようにテーブルやビューが追加されていることが確認できます。
最後に、サンプルプロジェクトのモデルを DWH へ生成します。
dbt run
以上で Elementary のセットアップは完了です。
Elementary テストの追加
Elementary で利用できる異常値検出の設定を行います。 チュートリアルでは以下の 3 つのモデルを対象とします。
customersordersreturned orders
まず、 customers には行数の異常値検出（row_count）を追加します。
models/schema.yml にある models.customers ブロックを以下のように編集します。
models: - name: customers description: This table has basic information about a customer, as well as some derived facts based on a customer's orders config: tags: ["PII"] elementary: timestamp_column: "signup_date" tests: - elementary.table_anomalies: table_anomalies: - row_count
このテストでは特定のタイムスタンプ列に対して行数をカウントし、平均に対して（3 × 標準偏差）だけ上回るか下回った日にちがあった場合に失敗する内容となっています。このケースでは signup_date 列ごとに顧客数合計を算出しています。
上記の例ではモデル単位での設定となっていますが、モデル内の特定の列に対してもテストを組み込むことができます。以下の例は customers の returned_orders 列に対して「ある期間に返送された注文数の異常値」を検出します。
- name: returned_orders description: This table contains all of the returned orders config: tags: ["finance"] tests: - elementary.table_anomalies: tags: ["table_anomalies"] table_anomalies: - row_count timestamp_column: "order_date"
次に、orders モデルにディメンションの値ごとに特定のフィールドの持つデータ分布に異常がないかをチェックします。
このテストでは status 列の値ごとに行数をカウントし、分布が平均から大きくずれている場合に失敗するようになっています。
- name: orders description: This table has basic information about orders, as well as some derived facts based on payments config: tags: ["finance"] elementary: timestamp_column: "order_date" tests: - elementary.dimension_anomalies: dimensions: - status
最後にcustomersモデルにあるnumber_of_orders列にオーダー数が 0 の日がないかを検出するテストを追加します。テストとしてzero_countをelementary.column_anomaliesに定義します。
- name: number_of_orders description: Count of the number of orders a customer has placed tests: - elementary.column_anomalies: config: severity: warn tags: ["column_anomalies"] column_anomalies: - zero_count timestamp_column: "signup_date"
以上でテスト設定は終了です。
テストの実行とレポートの生成
最後にテストとレポートを作成していきましょう。
Elementary では dbt とは CLI を使ってレポーティングを実行します。dbt とは別途接続情報の設定が必要ですので、下記コマンドを使って Elemantary のプロファイルを作成します。出力された内容は profiles/profiles.yml に追記します。
dbt run-operation elementary.generate_elementary_cli_profile
なお、サンプルリポジトリでは環境変数を使ったプロファイルを別途追記していますので上記の出力内容を利用する必要はありません。
dbt と同様に env_var を使った環境変数の呼び出しに対応しているようです。
elementary: outputs: default: type: redshift host: "{{ env_var('REDSHIFT_HOST') }}" port: "{{ env_var('REDSHIFT_PORT') | as_number }}" user: "{{ env_var('REDSHIFT_USERNAME') }}" password: "{{ env_var('REDSHIFT_PASSWORD') }}" dbname: "{{ env_var('REDSHIFT_DBNAME') }}" schema: "{{ env_var('REDSHIFT_SCHEMA') + '_elementary' }}" threads: 4
まず異常値を含むデータを元にモデルを再生成します。
dbt run --vars "{'anomalies': True}"
次にテストを実行します。
dbt test
schema.yml に定義したすべてのテストが失敗し、エラー3 件と警告 1 件が出力されます。
10:28:55 Finished running 4 tests, 2 hooks in 0 hours 0 minutes and 54.35 seconds (54.35s). 10:28:55 10:28:55 Completed with 3 errors and 1 warning: 〜〜（中略）〜〜 10:28:55 Done. PASS=0 WARN=1 ERROR=3 SKIP=0 TOTAL=4
Elementary レポートを生成します。 レポートは dbt docs と同様に、静的 HTML ファイルとして生成されます。
# プロファイルを含むディレクトリを指定する必要があるので注意 edr monitor report --profiles-dir $DBT_PROFILES_DIR
実際にレポートの内容を確認してみます。
トップページ（Test Results）では最新のテスト結果が一覧表示されます。
デフォルトでは過去 7 日間のテスト結果が表示されます（CLI の --days-back オプションで変更できます）。
テスト結果は Elementary の異常値テストのほかに dbt テストやスキーマ変更のテストも表示対象となっているようです。
テストごとに詳細を確認できます。異常値と期待値の比較がグラフ化され把握がしやすいですね。
Test Runs 画面では、デフォルトで過去 7 日間に実行された直近 30 回分のテスト実行履歴や失敗頻度などが確認できます。
Model Runs 画面では、dbt runでの結果履歴が確認できます。
実行時間にボトルネックのあるモデルがないかの発見に役立ちそうです。
Lineage では上流・下流のモデル間の依存関係を可視化することができます。 従来のdbt docs にも同様の機能がありますが、こちらはテスト結果とともに確認ができるので、異常値検出の根本的な原因や下流でモデルを利用しているユーザーやアプリケーションなどの影響範囲を把握するのに便利です。
Slack通知を試してみる
Elementary では、Slack と連携してレポートサマリやアラートをチャンネルに投稿する機能もあります。 連携方法には Token と Webhook の 2 通りあり、Webhook ではテストレポートが投稿できないなどいくつか差異があるようですので用途に応じて検討してみてください。 今回は Token を使ってアラートを発出させてみます。Slackbot や Slack Token の設定方法についてはドキュメントを参考にしてください。
Slack へのチャンネル投稿は次のコマンドで実行します。
edr monitor --slack-token $SLACK_TOKEN --slack-channel-name $SLACK_CHANNEL_NAME
今回のチュートリアルを使ったケースでは、モニタ終了後に指定の Slack チャンネルへアラートが 4 件（エラー3 件・警告 1 件）投稿されます。詳細表示からエラーメッセージやテスト時のパラメータが確認できるようです。
Token を使った方法では先程生成したレポートをチャンネルに直接送信することもできます。レポート送信は次のコマンド（edr monitor send-report）で行います。
edr monitor send-report --slack-token $SLACK_TOKEN --slack-channel-name $SLACK_CHANNEL_NAME
Slack 以外にも S3 バケットや GCS バケットへもレポートを送信できますので、サイトホスティングも容易ですね。
さいごに
今回はデータ品質モニタリングツールの Elementary を試してみました。dbt で作成したモデルに対する異常値検出テストや、レポーティング生成と Slack へのアラートまで容易に実現できるほか、dbt run実行時のボトルネックを見つけることができます。
ここでは紹介しきれていませんが、カスタムテストを Python を使って実装できるようで、そちらも個人的に注目しています。
データ分析では対象となるデータの量だけでなく、その精度や正しさが価値を生み出します。今後は Elementary のようなモニタリングツールを使ってデータ品質の可視化や適切なアクション・改善といった一連のサイクルを取り入れていきたいですね！
最後まで読んでいただきありがとうございました。
