Snowflake のデータメトリック関数で、毎朝取り込んだテーブルを自動で確かめる｜期待値で合否を決め、違反の日だけメールで知らせる

毎朝取り込んでいる注文テーブルに同じ注文番号の行が二重に入っていても、集計の数字が合わないと言われるまで気づけないことがあります。チェックの SELECT を毎朝自分で実行するのは手間です。自動のチェックにしても、問題の日に知らせが来なければ役に立ちません。

Snowflake には、テーブルの NULL・重複・件数・鮮度を測るデータメトリック関数（DMF）があります。DMF をテーブルに関連付け、合否の基準を期待値で書いておくと、決めた時刻に値と合否が自動で残ります。そこにアラートを足すと、違反があった日だけメールで知らせられます。SQL は、試すための文と、本番に入れる一式に分けて載せました。

ここに載せる SQL と結果の画面は、データ基盤の構築を支援している当社が、試用の Snowflake で架空の注文データに対して実行したものです。

DMF と期待値で毎回の値と合否を残し、違反した日だけアラートで知らせる

DMF は、列の NULL の数や重複の数のような、データの性質を1つの値で返す関数です。その値が良いか悪いかは、DMF そのものは決めません。合否を決めるのは期待値で、DMF と期待値の組み合わせを、Snowflake はデータ品質チェックと呼んでいます。使えるのは Enterprise Edition 以上です。

今回は、注文テーブル ORDERS に DMF を5つ関連付けました。金額の NULL と注文番号の重複を数え、取り込みの鮮度と、予定の時刻の前24時間に入った件数を測ります。この4つには期待値を付けました。残りの ROW_COUNT は、テーブル全体の件数を記録するだけです。

DMF の値は、テーブルごとに決めた DMF のスケジュールの時刻に、自動で測られます。測った値と期待値の合否は SNOWFLAKE データベースの2つのビューに残るので、記録用のテーブルを自分で作る必要はありません。

違反を知らせるのはアラートです。アラートは、決めた時刻に条件の SQL を実行し、結果が1行以上あれば処理を動かす Snowflake の仕組みです。今回は、毎朝6時40分に期待値を評価し、違反があればメールを送るアラートを作りました。

このあと本文に出てくる SQL は、検証用のデータベース TEST_DQ の中で、問題の行をわざと入れて確かめるためのものです。本番のテーブルに入れる一式は、後半に分けて載せました。

TEST_DQ に100行の注文テーブルを作り、DMF を SELECT で呼ぶ

先に1つ断っておきます。Snowflake のドキュメントのデータ品質の紹介ページには、「トライアルアカウントはこの機能をサポートしていません」とあります。一方、トライアルアカウントのページにある制限の一覧には、データ品質は挙がっていません。

今回の試用の環境では、試した操作はすべて動きました。それでも、試用の環境で試せる前提にはせず、自社の Enterprise Edition 以上のアカウントで確かめてください。

試したのは2026年10月21日の未明で、時刻はすべて日本時間です。環境は AWS の東京リージョンの試用の Snowflake で、ロールは ACCOUNTADMIN、ウェアハウスは X-Small にしました。

まず、検証用のデータベース TEST_DQ に注文テーブル ORDERS を作り、GENERATOR 関数で作った100行を入れます。名前は大文字にしています。あとの SELECT の WHERE で、データベース名を文字列として比べるからです。自分の名前に替えるときも、WHERE の ‘…’ の中は大文字で書きます。

USE ROLE ACCOUNTADMIN;
USE SECONDARY ROLES NONE;
USE WAREHOUSE COMPUTE_WH;
CREATE DATABASE IF NOT EXISTS TEST_DQ;
CREATE SCHEMA IF NOT EXISTS TEST_DQ.RAW;
USE SCHEMA TEST_DQ.RAW;
CREATE OR REPLACE TABLE ORDERS (ORDER_ID NUMBER, ORDER_DATE DATE, STORE STRING, AMOUNT NUMBER, LOADED_AT TIMESTAMP_LTZ);
INSERT INTO ORDERS SELECT SEQ4() + 1, '2026-10-20'::DATE, DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'), 1000 + MOD((SEQ4() + 1) * 7919, 9000), CURRENT_TIMESTAMP() FROM TABLE(GENERATOR(ROWCOUNT => 100));

LOADED_AT は、行を取り込んだ時刻の列です。型を TIMESTAMP_LTZ にしたのは、鮮度を測る FRESHNESS に渡すためです。ドキュメントで FRESHNESS の対象に挙がっている型は、DATE・TIMESTAMP_LTZ・TIMESTAMP_TZ の3つです。

自社のテーブルの時刻の列が TIMESTAMP_NTZ なら、渡せるかを先に確かめてください。列を渡さずに ON () とすると、FRESHNESS は最後に行が変わった時刻から測ります。

USE SECONDARY ROLES NONE は、USE ROLE で選んだロールの権限だけで試すための文です。理由は、マスキングポリシーの記事に書きました。

DMF は、SELECT の中で手で呼ぶこともできます。

USE SCHEMA TEST_DQ.RAW;
SELECT SNOWFLAKE.CORE.NULL_COUNT(SELECT AMOUNT FROM ORDERS) AS NULL_AMOUNT,
       SNOWFLAKE.CORE.DUPLICATE_COUNT(SELECT ORDER_ID FROM ORDERS) AS DUP_ORDER_ID;

結果は、NULL_AMOUNT も DUP_ORDER_ID も0でした。手で呼んだ分には DMF の料金はかからず、SELECT を実行したウェアハウスの料金だけがかかります。

テーブル全体の件数を測る ROW_COUNT は、手では呼べません。上の SELECT に、SNOWFLAKE.CORE.ROW_COUNT(SELECT ORDER_ID FROM ORDERS) を足しました。結果は次のエラーで、前後と引数の中は略しています。

too many arguments for function [ROW_COUNT$V1(...)] expected 0, got 1

ROW_COUNT は、テーブルに関連付けるときに、列を指定せず ON () と書きます。ドキュメントにも、この関数は直接呼べないとあります。

テーブルに関連付けて、DMF のスケジュールと期待値を決める

DMF は、ALTER TABLE でテーブルの列に関連付けます。測る時刻は、テーブルの DATA_METRIC_SCHEDULE で決めます。試すときは、すぐ結果を見るために5分ごとにしました。

USE SCHEMA TEST_DQ.RAW;
ALTER TABLE ORDERS SET DATA_METRIC_SCHEDULE = '5 MINUTE';
ALTER TABLE ORDERS ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.NULL_COUNT ON (AMOUNT);
ALTER TABLE ORDERS ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.DUPLICATE_COUNT ON (ORDER_ID);
ALTER TABLE ORDERS ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.ROW_COUNT ON ();
SELECT METRIC_NAME, REF_ENTITY_NAME, SCHEDULE, SCHEDULE_STATUS
FROM TABLE(INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS', REF_ENTITY_DOMAIN => 'TABLE'));

最後の SELECT は、関連付けの一覧です。3つとも SCHEDULE_STATUS が STARTED で、SCHEDULE は */5 * * * * UTC でした。5 MINUTE と書いても、一覧では cron の形になります。

DMF のスケジュールは、テーブルに1つです。関連付けた DMF は、どれも同じ時刻に測られます。分で書くなら、5・15・30・60・720・1440 のどれかです。cron で書くなら、間隔は5分以上にします。TRIGGER_ON_CHANGES にすると、行を足したり消したりしたときに測ります。

スケジュールを決めずに関連付けると、既定の60分ごとに測られます。毎朝1回のつもりが1日24回になるので、先に DATA_METRIC_SCHEDULE を決めてから関連付けます。

ドキュメントには、スケジュールを設定してから効くまでに10分かかるとあります。ただし、新しく関連付けた DMF には、この遅れはかからないとも書いてあります。今回は 01:58 ごろに設定と関連付けをして、予定の時刻が 02:00 の測定から値が残っていました。

次に、期待値を付けます。関連付け済みの DMF には、MODIFY … ADD EXPECTATION で付けます。新しく関連付けるときは、ADD の文の後ろに EXPECTATION を続けて書けます。

USE SCHEMA TEST_DQ.RAW;
ALTER TABLE ORDERS MODIFY DATA METRIC FUNCTION SNOWFLAKE.CORE.NULL_COUNT ON (AMOUNT)
  ADD EXPECTATION AMOUNT_NOT_NULL (VALUE = 0);
ALTER TABLE ORDERS MODIFY DATA METRIC FUNCTION SNOWFLAKE.CORE.DUPLICATE_COUNT ON (ORDER_ID)
  ADD EXPECTATION ORDER_ID_UNIQUE (VALUE = 0);
ALTER TABLE ORDERS ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.FRESHNESS ON (LOADED_AT)
  EXPECTATION LOADED_WITHIN_26H (VALUE < 93600);
ALTER TABLE ORDERS ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.ROWS_IN_PAST_DAY_COUNT ON (LOADED_AT)
  EXPECTATION AT_LEAST_50_ROWS (VALUE >= 50);

期待値の式では、DMF が返した値を VALUE と書き、比較演算子で基準を書きます。式が TRUE なら合格、FALSE なら違反です。式からは、ほかのテーブルやビューを参照できません。前日の値と比べるような判定は、期待値では書けません。

NULL_COUNT と DUPLICATE_COUNT は、テーブル全体の行を数えます。過去の行に NULL や重複が1件でも残っていると、VALUE = 0 の期待値は、直すまで毎朝違反になります。始める前に今の値を確かめる手順は、最後の節に書きました。評価する行を条件で絞る書き方もドキュメントにありますが、今回は試していません。

DUPLICATE_COUNT は、NULL も重複として数えます。注文番号に NULL が入りうるなら、その列に NULL_COUNT も関連付けて、分けて見ます。

鮮度の上限は、取り込みの時刻と評価の時刻から決めます。たとえば、毎朝6時までに取り込みが終わり、アラートが6時40分に評価する場合です。今朝の取り込みが来ていれば鮮度は40分以上で、来ていなければ前の日の取り込みから24時間40分ほどです。

上限を短くすると、取り込みが少し遅れた朝も違反になります。長くすると、取り込みが来なかった朝も合格になります。上限は、ふだんの取り込みの遅れを見込んだうえで、24時間40分より短くします。

今回の上限の93600秒、つまり26時間は、試すときの値です。6時の取り込みと6時40分の評価の組み合わせでは、今朝の取り込みが来なくても合格になります。ただし、その朝は件数の期待値が違反になります。前の日の6時の行は、6時40分の24時間前より前なので数えないからです。どちらも計算で、試してはいません。

ROWS_IN_PAST_DAY_COUNT は、時刻の列の値が、予定の時刻の前24時間に入っている行を数えます。暦の1日の件数ではありません。下限の50行も試すときの値で、今回の1日分の100行の半分にしました。本番では、ふだんの件数から決めます。

問題の行を入れ、期待値を評価して違反を確かめる

DMF のスケジュールを待たずに合否を見るときは、SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS を使います。テーブルの期待値を、その場で評価する関数です。使わない列が多いので、EXCLUDE で外しています。

USE SCHEMA TEST_DQ.RAW;
SELECT * EXCLUDE (METRIC_DATABASE, METRIC_SCHEMA, EXPECTATION_ID, ARGUMENTS)
FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS'));

問題を入れる前は、4つの期待値がすべて合格でした。FRESHNESS の値は秒で、199なら、最後に取り込んでから199秒という意味です。

次に、わざと問題の行を入れます。入れるのは、注文番号1・2・3と同じ番号の3行と、金額が NULL の2行で、合わせて105行になります。

USE SCHEMA TEST_DQ.RAW;
INSERT INTO ORDERS VALUES
  (1, '2026-10-20', '渋谷店', 3000, CURRENT_TIMESTAMP()),
  (2, '2026-10-20', '新宿店', 4000, CURRENT_TIMESTAMP()),
  (3, '2026-10-20', '渋谷店', 5000, CURRENT_TIMESTAMP()),
  (201, '2026-10-20', '新宿店', NULL, CURRENT_TIMESTAMP()),
  (202, '2026-10-20', '渋谷店', NULL, CURRENT_TIMESTAMP());
SELECT * EXCLUDE (METRIC_DATABASE, METRIC_SCHEMA, EXPECTATION_ID, ARGUMENTS, GROUP_BY_VALUES, _CACHE_BUSTER)
FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS'));

入れたあとに評価すると、NULL_COUNT が2、DUPLICATE_COUNT が3になり、この2つの期待値が違反になりました。鮮度と件数の期待値は合格のままです。2と3は、105行のテーブル全体を数えた値です。

DUPLICATE_COUNT の3は、重複に関わる6行の数ではありませんでした。今回はどの番号も2行ずつなので、重複がある値の数と、余分に入った行の数がどちらも3です。同じ番号が3行以上あるときの数え方は試していません。

違反が分かったら、該当の行を取り出します。SYSTEM$DATA_METRIC_SCAN に、テーブルと DMF の名前と列を渡すと、その DMF が数えた行が返ります。

USE SCHEMA TEST_DQ.RAW;
SELECT ORDER_ID, ORDER_DATE, STORE, AMOUNT
FROM TABLE(SYSTEM$DATA_METRIC_SCAN(
  REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS',
  METRIC_NAME => 'snowflake.core.duplicate_count',
  ARGUMENT_NAME => 'ORDER_ID'))
ORDER BY ORDER_ID, AMOUNT;

返ったのは6行で、注文番号1・2・3のそれぞれに、元の行と足した行がありました。どちらを残すかは、この結果だけでは決められません。取り込み元のデータと見比べてから直します。

金額が NULL の行なら、METRIC_NAME を ‘snowflake.core.null_count’ にします。ARGUMENT_NAME は ‘AMOUNT’ です。こちらは試していません。直したら、期待値をもう一度評価して、FALSE に戻ったかを見ます。直すまでは、アラートが毎朝メールを送ります。

DMF のスケジュールで測った値と合否は、2つのビューに残る

DMF のスケジュールで測った値は、SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_RESULTS に残ります。この記事では、結果のビューと書きます。

USE SCHEMA TEST_DQ.RAW;
SELECT METRIC_NAME, VALUE,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'HH24:MI:SS') AS SCHEDULED_JST,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', MEASUREMENT_TIME), 'HH24:MI:SS') AS MEASURED_JST
FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_RESULTS
WHERE TABLE_DATABASE = 'TEST_DQ' AND TABLE_NAME = 'ORDERS'
ORDER BY MEASUREMENT_TIME DESC, METRIC_NAME
LIMIT 20;

02:00 の測定は、5つの DMF がどれも 02:00:15 に測っていました。問題を入れる前の測定で、NULL_COUNT と DUPLICATE_COUNT は0、FRESHNESS は192です。02:05 の測定では NULL_COUNT が2、DUPLICATE_COUNT が3になり、2つの件数は105に増えました。

02:05 の測定は、02:05:14 から 02:05:50 にかけて行われ、02:06 ごろには結果のビューで見られました。

測った順に並べるときは、予定の時刻の SCHEDULED_TIME ではなく、測った時刻の MEASUREMENT_TIME を使います。予定の時刻と測った時刻の間にも、行が入ることがあるからです。

期待値の合否は、SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS に残ります。こちらは合否のビューと書きます。

USE SCHEMA TEST_DQ.RAW;
SELECT * EXCLUDE (TABLE_DATABASE, TABLE_SCHEMA)
FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_EXPECTATION_STATUS
WHERE TABLE_DATABASE = 'TEST_DQ' AND TABLE_NAME = 'ORDERS'
ORDER BY MEASUREMENT_TIME DESC
LIMIT 10;

合否のビューは、4つの期待値の2回分で8行でした。違反を示す TRUE だったのは、02:05 の測定の ORDER_ID_UNIQUE と AMOUNT_NOT_NULL です。

2つのビューを見るには、アプリケーションロールを GRANT APPLICATION ROLE で付与します。見るだけなら、SNOWFLAKE.DATA_QUALITY_MONITORING_VIEWER です。データベースロールの SNOWFLAKE.GOVERNANCE_VIEWER では、結果のビューは見られないとドキュメントにあります。

アラートの条件では、EXPECTATION_VIOLATED を ‘TRUE’ と文字列で比べる

メールの宛先にできるのは、同じアカウントの Snowflake ユーザーで、メールアドレスを確認済みの人だけです。送るには、通知インテグレーションも要ります。Snowflake からメールなどで外へ知らせるための設定で、送ってよい宛先を ALLOWED_RECIPIENTS に書きます。

試すときは、受け取り用のユーザーと通知インテグレーションを、2つの SQL に分けて作ります。名前は、タスクで取り込みを自動にする記事とそろえました。その記事の片付けまで済ませた場合、この2つは消えているので、ここで作り直します。

次の2つの塊は、その記事の試しで実行した文です。ユーザーの COMMENT の文は、試したときの文から置き換えています。まず、受け取り用のユーザーを作り、確認メールを送ります。

USE ROLE ACCOUNTADMIN;
CREATE USER IF NOT EXISTS DATA_TEAM_NOTIFY
  TYPE = PERSON
  EMAIL = 'data-team@example.com'
  COMMENT = 'email recipient only, no password';
CALL SYSTEM$START_USER_EMAIL_VERIFICATION('DATA_TEAM_NOTIFY');

受け取る人が確認メールのリンクを開き、アドレスが確認済みになってから、次を実行します。確認していないアドレスを ALLOWED_RECIPIENTS に入れると、CREATE が失敗するとドキュメントにあります。ユーザー自身が Snowsight で確認することもできます。

USE ROLE ACCOUNTADMIN;
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS LOAD_EMAIL_INT
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('data-team@example.com');

宛先の data-team@example.com は説明用のアドレスなので、自社の確認済みのアドレスに替えます。IF NOT EXISTS なので、同じ名前の通知インテグレーションがすでにあると、何も作られず、宛先も足されません。ALLOWED_RECIPIENTS に入っていない宛先が1つでもあると、メールは送られないとドキュメントにあります。すでにある場合は、宛先が入っているかを確かめます。

メールを送るのは、ストアドプロシージャの NOTIFY_DQ_VIOLATIONS です。ストアドプロシージャは、SQL で書いた処理の手順を保存したものです。期待値を評価し、違反した期待値の数と一覧を本文にして、SYSTEM$SEND_EMAIL で送ります。

アラート CHECK_DQ は、毎朝6時40分に同じ関数で期待値を評価し、違反した期待値が1つでもあれば、このストアドプロシージャを呼びます。最後の EXECUTE ALERT は、予定の時刻を待たずに一度だけ実行する文です。件名と本文は、試したときのままです。

USE SCHEMA TEST_DQ.RAW;
CREATE OR REPLACE PROCEDURE NOTIFY_DQ_VIOLATIONS()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  n NUMBER;
  detail STRING;
  body STRING;
BEGIN
  SELECT COUNT(*), LISTAGG(METRIC_NAME || ' / ' || EXPECTATION_NAME || ' の値 ' || VALUE, ' | ')
    INTO :n, :detail
    FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS'))
    WHERE EXPECTATION_VIOLATED = 'TRUE';
  body := 'これはブログ記事の検証用のテスト送信です。' || CHR(10) ||
          '注文テーブルのデータ品質の期待値に違反があります: ' || n || ' 件' || CHR(10) ||
          COALESCE(detail, '');
  CALL SYSTEM$SEND_EMAIL(
    'LOAD_EMAIL_INT',
    'data-team@example.com',
    '[テスト送信] Snowflake データ品質の違反（ブログ検証）',
    :body);
  RETURN n;
END;
$$;
CREATE OR REPLACE ALERT CHECK_DQ
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 40 6 * * * Asia/Tokyo'
  IF (EXISTS (
    SELECT 1 FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS'))
    WHERE EXPECTATION_VIOLATED = 'TRUE'
  ))
  THEN
    CALL NOTIFY_DQ_VIOLATIONS();
EXECUTE ALERT CHECK_DQ;

最初は、条件を WHERE EXPECTATION_VIOLATED と書きました。違反なら TRUE が入る真偽値の列だと思ったからです。アラートもストアドプロシージャも、作るときはエラーになりませんでした。

ところが EXECUTE ALERT で実行すると、アラートの実行履歴の STATE は CONDITION_FAILED でした。SQL_ERROR_CODE は2034で、エラー文は次のとおりです。途中は略しています。

SQL compilation error: Invalid data type [VARCHAR(5)] for predicate [… EXPECTATIONS.EXPECTATION_VIOLATED]

期待値を評価する関数が返す EXPECTATION_VIOLATED は、真偽値ではなく VARCHAR(5) の文字列でした。’TRUE’ か ‘FALSE’ の文字が入ります。

写真で見比べると、列名の左に出る型のアイコンが違います。期待値を評価する関数の結果では、文字列を示す A でした。合否のビューでは、真偽値を示す 0|1 です。リファレンスの戻り値の一覧では、この関数の EXPECTATION_VIOLATED も BOOLEAN と書かれていて、今回の結果と食い違います。型が BOOLEAN に変わったときに = ‘TRUE’ の比較が通るかは、試していません。

条件を WHERE EXPECTATION_VIOLATED = 'TRUE' に直しました。実行すると、アラートの実行履歴は TRIGGERED になり、送信の記録の STATUS は SUCCESS でした。上の SQL は、直したあとの文です。

SUCCESS は、Snowflake がメールを送り出したという記録で、受信箱に届いたことまでは示しません。同じ通知インテグレーションで送ったメールは、取り込みを自動にする記事のときに、受信箱に届いたことを確かめています。今回のメールは、送信の記録までしか見ていません。

今回のように、作るときにエラーにならなくても、実行すると失敗することがあります。アラートを作ったら、一度 EXECUTE ALERT で実行し、アラートの実行履歴の STATE を見ます。違反がある状態で実行すると、そのときメールが送られます。

作った直後のアラートは、一時停止の状態です。毎朝の予定で動かすには、ALTER ALERT … RESUME を実行します。今回は RESUME していないので、6時40分に動くところは見ていません。

アラートの条件は、DMF のスケジュールで残った合否ではなく、アラートが動いた時点のテーブルを評価して決まります。そのため、鮮度のように時刻で変わる値は、ビューの記録とアラートの評価で一致しません。合否のビューを条件にする形も考えられますが、試していません。

CREATE OR REPLACE TABLE のあと、関連付けの一覧は0件だった

取り込みのたびに CREATE OR REPLACE TABLE でテーブルを作り直しているなら、気をつけることがあります。テーブルを同じ中身で作り直してから、関連付けの一覧を数えました。

USE SCHEMA TEST_DQ.RAW;
CREATE OR REPLACE TABLE ORDERS AS SELECT * FROM ORDERS;
SELECT COUNT(*) AS DMF_BINDINGS_AFTER_RECREATE
FROM TABLE(INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS', REF_ENTITY_DOMAIN => 'TABLE'));

結果は0件でした。ドキュメントには、作るときに DMF を関連付ける書き方の説明があります。その中に、CREATE OR REPLACE は関連付けを一から置き換える、とあります。

毎朝テーブルを作り直す取り込みでは、関連付けた DMF と期待値が毎朝外れ、何も測られなくなります。アラートが呼ぶ評価の関数も失敗する見込みで、違反がある日でもメールは来ないはずです。

作り直す取り込みのままにするなら、CREATE 文の中に WITH DATA METRIC FUNCTION を書く方法があります。作るときに、DMF と期待値を関連付ける書き方です。作り直したあとに DMF のスケジュールが残るかは、確かめていません。テーブルを作り直さず、TRUNCATE してから INSERT する形や、MERGE で入れ替える形もあります。WITH の書き方も、作り直さない形も、今回は試していません。

差分を MERGE で足していく組み方は、ストリームとタスクで集計テーブルを更新する記事に書きました。

知らせが来ない朝は、関連付けの件数とアラートの実行履歴で確かめる

アラートが鳴らなかった朝に、問題が無かったとは限りません。この一式が何も知らせなくなる朝が、少なくとも3つあるからです。

• テーブルを作り直して、関連付けが0件になった朝。
• アラートそのものが失敗した朝。今回の CONDITION_FAILED のように、メールは送られません。
• アラートを RESUME する前か、一時停止になったままの朝。アラートは動きません。

1つ目は、今のアラートでは拾えない見込みです。期待値を評価する関数の引数は、期待値の付いた DMF を1つ以上持つテーブル、とドキュメントにあります。関連付けが0件の朝はこの関数が失敗し、アラートは CONDITION_FAILED になると考えられます。

そこで、この関数を呼ばない別のアラートを置きます。条件は、関連付けの一覧の件数が0であることです。処理では、ストアドプロシージャを通さずにメールを送ります。次の SQL は、今回は実行していません。関連付けの一覧を数える関数の引数は、試したときの SELECT と同じです。

-- 関連付けが0件の朝に知らせる、もう1つのアラート（今回は実行していない）
CREATE OR REPLACE ALERT CHECK_DQ_BINDINGS
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <アラートの分> <アラートの時> * * * <タイムゾーン>'
  IF (EXISTS (
    SELECT 1
    FROM (SELECT COUNT(*) AS N
          FROM TABLE(<データベース>.INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(REF_ENTITY_NAME => '<データベース>.<スキーマ>.<テーブル>', REF_ENTITY_DOMAIN => 'TABLE')))
    WHERE N = 0
  ))
  THEN
    CALL SYSTEM$SEND_EMAIL(
      '<通知インテグレーション>',
      '<宛先>',
      '<関連付けが外れたときの件名>',
      '<テーブルの呼び名>に関連付けた DMF が0件です。期待値は評価されていません。');

INFORMATION_SCHEMA の前にデータベース名を付けたのは、アラートとテーブルが別のデータベースにあっても、テーブルの側を読むためです。このアラートも、作ったあとは一時停止の状態なので、EXECUTE ALERT で一度確かめてから RESUME します。

2つ目と3つ目は、アラートの中では拾えません。人が毎朝、次の2つの SELECT を1つずつ実行して見ます。関連付けの一覧が0件になっていないかと、アラートの実行履歴に今朝の行があるかです。

USE SCHEMA TEST_DQ.RAW;
SELECT METRIC_NAME, REF_ENTITY_NAME, SCHEDULE, SCHEDULE_STATUS
FROM TABLE(INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(REF_ENTITY_NAME => 'TEST_DQ.RAW.ORDERS', REF_ENTITY_DOMAIN => 'TABLE'));
SELECT 'ALERT' AS SRC, NAME AS VALUE1, STATE AS VALUE2, COALESCE(SQL_ERROR_MESSAGE, '') AS VALUE3,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'HH24:MI:SS') AS AT_JST
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(SCHEDULED_TIME_RANGE_START => DATEADD(MINUTE, -30, CURRENT_TIMESTAMP())))
WHERE NAME = 'CHECK_DQ'
UNION ALL
SELECT 'MAIL', '(送信の記録)', STATUS, '', TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', CREATED), 'HH24:MI:SS')
FROM TABLE(INFORMATION_SCHEMA.NOTIFICATION_HISTORY(START_TIME => DATEADD(MINUTE, -10, CURRENT_TIMESTAMP())))
ORDER BY AT_JST DESC;

SRC が ALERT の行はアラートの実行履歴、MAIL の行は送信の記録です。STATE が TRIGGERED なら違反があってメールを送った朝で、CONDITION_FALSE なら違反が無かった朝です。試したときは30分前と10分前からを見たので、毎朝見るときは、DATEADD の幅を前の朝まで届くように広げます。

アラートの実行履歴に今朝の行が無ければ、アラートが一時停止のままになっていないかを確かめます。CONDITION_FAILED か ACTION_FAILED の行があれば、エラー文を見て、条件か処理を直します。直したら、EXECUTE ALERT で一度実行します。

失敗した朝と一時停止の朝は、動かし始めて何日たっても起こりえます。そのため、この2つの SELECT は、動かし始めてからも続けて見ます。続けて失敗したアラートを自動で一時停止する SUSPEND_ALERT_AFTER_NUM_FAILURES という設定も、ドキュメントにあります。この設定で一時停止になった朝も、実行履歴に今朝の行が無いことで気づけます。

期待値で書けない判定と、料金のかかり方

期待値は、DMF の値を決まった値と比べることしかできません。件数が前日より大きく減った日、のような判定は書けません。

こうした判定には、2つの方法があります。1つは、結果のビューから前日の値を引いて比べる SQL を自分で書き、アラートの条件にする形です。もう1つは、DMF の異常検出です。

異常検出は、過去の値から予測した範囲を外れた日を見つける機能で、プレビューです。対象は、件数の ROW_COUNT と鮮度の FRESHNESS だけです。頻繁に測る DMF では、検出を始めるまでに少なくとも2週間分の値が要ります。60日分で学習させることが勧められています。

次の図は、自分で書く SQL と DMF を、書ける判定・記録・料金・作り直したときなどで比べたものです。

料金は、DMF のスケジュールで測るたびにかかります。DMF はサーバーレスで動き、請求では Data Quality Monitoring に入ります。結果を残す分は、Logging として別に請求に出ます。DMF を作るだけでは、料金はかかりません。

単価は、ドキュメントでは Service Consumption Table という資料を見るよう案内されています。日ごとの使用量を見るビューは、DATA_QUALITY_MONITORING_USAGE_HISTORY です。SNOWFLAKE.ACCOUNT_USAGE のスキーマにあり、反映は最大180分遅れます。

アラートは、違反の無い朝も毎朝条件を評価します。ウェアハウスを指定すると、メールを送るだけの処理でも、最低1分ぶんのウェアハウスの料金がかかります。Snowflake の料金がどこでかかるかは、料金とコストの記事にまとめています。

スケジュールは、細かくしすぎないようにします。今回の5分ごとは、すぐ結果を見るための試しの設定です。毎朝1回の取り込みなら、cron で1日1回にします。DMF のスケジュールでの評価をやめるときは、SET DATA_METRIC_SCHEDULE = '' にします。UNSET DATA_METRIC_SCHEDULE では評価は続き、既定の60分ごとに戻ります。

今回、試していないことと見ていないことは、次のとおりです。

• 前日比の SQL、異常検出、データベース単位の通知の設定
• Snowsight のデータ品質の画面、TRIGGER_ON_CHANGES、自作の DMF、評価する行を絞る書き方
• 合否のビューを条件にしたアラート、6時40分の予定で動くところ、2通目以降の受信
• 作り直したあとの DMF のスケジュール、鮮度と件数の期待値が違反になる日
• DMF の使用量と料金、権限を絞ったロールでの運用

読んだドキュメントからは、結果のビューに値が出るまでの時間と、何日分残るかは分かりませんでした。アラートの中で期待値を評価した分が、DMF の料金に入るかどうかも分かりませんでした。

本番のテーブルに入れる一式は、試した文から本番に要る文だけを残す

本番のテーブルに入れるときは、次の一式を上から順に実行します。最後の RESUME だけは、アラートを一度確かめてからにします。わざと問題を入れる INSERT と、予定の時刻を待たずに実行する EXECUTE ALERT は外しました。DMF のスケジュールは、5分ごとから毎朝の時刻に変えています。

<…> は、本番の値に書き換える所です。

• <データベース>.<スキーマ>.<テーブル> と <ウェアハウス> は、本番のテーブルとウェアハウスの名前にします。REF_ENTITY_NAME の文字列の中にもあります。
• <NULL を許さない列>・<重複を許さない列>・<取り込んだ時刻の列> は、自社の列名にします。
• <NULL の期待値> などの4つは期待値の名前で、<鮮度の上限の秒数> と <件数の下限> は基準です。どちらも、基準の決め方に合わせて付けます。
• <DMFの分> <DMFの時> は取り込みが終わったあとに、<アラートの分> <アラートの時> はそれより後にします。<タイムゾーン> は、両方の cron に使います。
• <通知インテグレーション>・<宛先>・<件名>・<本文の書き出し>・<テーブルの呼び名> は、本番で送るメールの内容にします。

-- 本番用。試した文のうち本番に要るものを残し、名前・列・基準・時刻・宛先を <…> にした（この形のまとまりでは実行していない）
USE ROLE ACCOUNTADMIN;
USE WAREHOUSE <ウェアハウス>;
USE SCHEMA <データベース>.<スキーマ>;
-- DMF のスケジュール（試したときは '5 MINUTE'。この cron の形は実行していない）
ALTER TABLE <テーブル> SET DATA_METRIC_SCHEDULE = 'USING CRON <DMFの分> <DMFの時> * * * <タイムゾーン>';
-- 5つの DMF と4つの期待値
ALTER TABLE <テーブル> ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.NULL_COUNT ON (<NULL を許さない列>);
ALTER TABLE <テーブル> ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.DUPLICATE_COUNT ON (<重複を許さない列>);
ALTER TABLE <テーブル> ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.ROW_COUNT ON ();
ALTER TABLE <テーブル> MODIFY DATA METRIC FUNCTION SNOWFLAKE.CORE.NULL_COUNT ON (<NULL を許さない列>)
  ADD EXPECTATION <NULL の期待値> (VALUE = 0);
ALTER TABLE <テーブル> MODIFY DATA METRIC FUNCTION SNOWFLAKE.CORE.DUPLICATE_COUNT ON (<重複を許さない列>)
  ADD EXPECTATION <重複の期待値> (VALUE = 0);
ALTER TABLE <テーブル> ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.FRESHNESS ON (<取り込んだ時刻の列>)
  EXPECTATION <鮮度の期待値> (VALUE < <鮮度の上限の秒数>);
ALTER TABLE <テーブル> ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.ROWS_IN_PAST_DAY_COUNT ON (<取り込んだ時刻の列>)
  EXPECTATION <件数の期待値> (VALUE >= <件数の下限>);
-- SCHEDULE_STATUS が STARTED になったか
SELECT METRIC_NAME, REF_ENTITY_NAME, SCHEDULE, SCHEDULE_STATUS
FROM TABLE(INFORMATION_SCHEMA.DATA_METRIC_FUNCTION_REFERENCES(REF_ENTITY_NAME => '<データベース>.<スキーマ>.<テーブル>', REF_ENTITY_DOMAIN => 'TABLE'));
-- 通知インテグレーション。同じ名前がすでにあると作られず、宛先も足されない
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS <通知インテグレーション>
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('<宛先>');
-- 違反を知らせるストアドプロシージャ
CREATE OR REPLACE PROCEDURE NOTIFY_DQ_VIOLATIONS()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  n NUMBER;
  detail STRING;
  body STRING;
BEGIN
  SELECT COUNT(*), LISTAGG(METRIC_NAME || ' / ' || EXPECTATION_NAME || ' の値 ' || VALUE, ' | ')
    INTO :n, :detail
    FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => '<データベース>.<スキーマ>.<テーブル>'))
    WHERE EXPECTATION_VIOLATED = 'TRUE';
  body := '<本文の書き出し>' || CHR(10) ||
          '<テーブルの呼び名>のデータ品質の期待値に違反があります: ' || n || ' 件' || CHR(10) ||
          COALESCE(detail, '');
  CALL SYSTEM$SEND_EMAIL(
    '<通知インテグレーション>',
    '<宛先>',
    '<件名>',
    :body);
  RETURN n;
END;
$$;
-- アラート
CREATE OR REPLACE ALERT CHECK_DQ
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <アラートの分> <アラートの時> * * * <タイムゾーン>'
  IF (EXISTS (
    SELECT 1 FROM TABLE(SYSTEM$EVALUATE_DATA_QUALITY_EXPECTATIONS(REF_ENTITY_NAME => '<データベース>.<スキーマ>.<テーブル>'))
    WHERE EXPECTATION_VIOLATED = 'TRUE'
  ))
  THEN
    CALL NOTIFY_DQ_VIOLATIONS();
-- 毎朝の予定で動かし始める。先に EXECUTE ALERT CHECK_DQ; で STATE を確かめてから実行する（今回は実行していない）
ALTER ALERT CHECK_DQ RESUME;

実行したら、関連付けの一覧で SCHEDULE_STATUS が STARTED かを見ます。最後の RESUME の前に、EXECUTE ALERT CHECK_DQ; を一度実行し、アラートの実行履歴の STATE を見ます。そのとき違反があれば、メールが送られます。関連付けが0件の朝にも知らせたいなら、知らせが来ない朝の節の、関連付けを数えるアラートを足します。

今回は、どれも ACCOUNTADMIN で試しました。ほかのロールで組むときに要る権限は、ドキュメントでは次のとおりです。

• DMF を関連付けるには、アカウントの EXECUTE DATA METRIC FUNCTION と、テーブルの OWNERSHIP が要ります。DMF のスケジュールを設定するのも、テーブルの OWNERSHIP を持つロールです。
• 2つのビューを見るロールには、SNOWFLAKE.DATA_QUALITY_MONITORING_VIEWER を付与します。
• アラートを作るロールには、アカウントの EXECUTE ALERT が要り、付与できるのは ACCOUNTADMIN のロールだけです。スキーマの USAGE と CREATE ALERT、ウェアハウスの USAGE も要ります。
• アラートとストアドプロシージャは、所有するロールの権限で動きます。そのロールには、期待値を評価するためのテーブルの SELECT と、通知インテグレーションの USAGE を付与します。
• 通知インテグレーションを作るには、CREATE NOTIFICATION INTEGRATION の権限が要ります。CREATE INTEGRATION でも作れます。

TEST_DQ のほかに、データベースの外に作った2つも消す

試し終えたら、アラートを一時停止してから TEST_DQ を消します。データ保持期間を0にしてから消すのは、保持期間の分のストレージ料金を残さないためです。データ保持期間は、Time Travel と UNDROP の記事で説明しています。

DMF の関連付けはテーブルに付いているので、テーブルと一緒に消え、別に外す必要はありません。受け取り用のユーザーと通知インテグレーションはデータベースの外にあるので、DROP で消します。本番でも使うなら、この2行は実行しません。

USE ROLE ACCOUNTADMIN;
ALTER ALERT TEST_DQ.RAW.CHECK_DQ SUSPEND;
ALTER DATABASE TEST_DQ SET DATA_RETENTION_TIME_IN_DAYS = 0;
DROP DATABASE TEST_DQ;
-- 下の2行は、本番でも使うなら実行しない
DROP USER IF EXISTS DATA_TEAM_NOTIFY;
DROP INTEGRATION IF EXISTS LOAD_EMAIL_INT;

消したあと、TEST_DQ、受け取り用のユーザー、通知インテグレーションが残っていないことを、それぞれの件数が0になったことで確かめました。

期待値を付ける前に、今のテーブルの NULL と重複を数える

期待値を付けた翌朝から違反のメールが続くかどうかは、今のテーブルの NULL と重複の数で決まります。毎朝取り込んでいるテーブルで、NULL_COUNT と DUPLICATE_COUNT を SELECT の中で呼んでみてください。呼ぶのは、NULL を許さない列と、重複を許さない列です。手で呼ぶ分は、DMF の料金がかかりません。

0なら、VALUE = 0 の期待値をそのまま付けられます。0でなければ、期待値を付ける前に過去の行を直すか、VALUE の基準を今の値に合わせて決めます。

dbt は、SQL で書いた変換をまとめて管理し、実行するツールです。dbt のテストは、取り込んだテーブルにも変換したテーブルにも書けて、dbt を実行したときに確かめます。DMF は、Snowflake の中で、決めた時刻にテーブルを測ります。当社は Snowflake 導入支援の中で、dbt のテストの整備をお手伝いしています。