Snowflake のタスクで毎朝の取り込みを自動にする｜0件の日とファイルを飛ばした日に気づく記録テーブルとアラート

毎朝届く注文の CSV を、手で COPY INTO して Snowflake に取り込んでいるとします。この作業をタスクで自動にすると、困るのは取り込めなかった日です。タスクの実行履歴が成功のままでも、新しいデータが1件も入っていない日があり、それに気づかないまま朝の集計が進むおそれがあります。

この記事では、取り込みを自動にするときに作る4つを、SQL つきで書きます。記録テーブル、取り込みに使うストアドプロシージャ（SQL で書いた処理の手順を保存したもの）、タスク、アラートです。1回ごとの結果が記録テーブルに1行ずつ残り、アラートがそれを毎朝見て、問題がある日だけメールで知らせます。

この記事の SQL と画面は、どれも試用の Snowflake で当社が動かして確かめたもので、注文の CSV は架空のデータです。

タスクのほかに、記録テーブル・ストアドプロシージャ・アラートを作る

タスクは、決めた時刻に SQL を実行する Snowflake の仕組みです。Snowflake の導入を支援している当社は、この一式を組む前に、タスクに COPY INTO を1文だけ書いて試しました。そのとき、困ったことが2つありました。

1つ目は、新しいファイルが無い日も、タスクの実行履歴が SUCCEEDED になったことです。どのファイルを取り込んだかは、ロードのメタデータとしてテーブルごとに記録され、取り込み済みのファイルは二度取り込まれません。何も入らなくても、タスクは成功になります。

2つ目は、形式の違うファイルが1つ混ざると、正しいファイルの120行も入らなかったことです。ON_ERROR を指定しないと、既定の ABORT_STATEMENT になり、COPY INTO の文全体が失敗します。この回は、取り込みの履歴にも行が残りませんでした。

どちらの場合も、タスクの実行履歴を見ただけでは、何行入ったかは分かりません。そこで、タスクのほかに3つを一緒に作ります。

ファイルを置く場所は、LANDING という内部ステージです。内部ステージは、Snowflake の中に用意するファイルの置き場所です。S3 などのクラウドのストレージを、外部ステージとして使う形もあります。毎朝ステージにファイルを置く仕組みは、この一式とは別に用意してください。

本文に載せる SQL は、どれも試すときの文です。試すときは、検証用のデータベース test_ingest のまま使います。本番に入れるときは、後半の「本番に入れる SQL」の節にまとめた一式を使い、その中の <…> を自社の名前にします。

試す土台は、次の SQL で作ります。検証用のデータベースとスキーマ、ステージ、取り込み先の ORDERS テーブルを作り、19日分の注文のファイルを書き出します。ファイルの中身は、行を作る関数 GENERATOR の結果です。

USE ROLE ACCOUNTADMIN;
USE WAREHOUSE COMPUTE_WH;
CREATE DATABASE IF NOT EXISTS test_ingest;
CREATE SCHEMA IF NOT EXISTS test_ingest.raw;
USE SCHEMA test_ingest.raw;
CREATE OR REPLACE STAGE LANDING;
COPY INTO @LANDING/orders/orders_2026-10-19.csv
  FROM (SELECT SEQ4() + 1 AS ORDER_ID, '2026-10-19'::DATE AS ORDER_DATE,
               DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店') AS STORE,
               1000 + MOD((SEQ4() + 1) * 7919, 9000) AS AMOUNT
        FROM TABLE(GENERATOR(ROWCOUNT => 100)))
  FILE_FORMAT = (TYPE = CSV COMPRESSION = NONE) HEADER = TRUE SINGLE = TRUE OVERWRITE = TRUE;
CREATE OR REPLACE TABLE ORDERS (ORDER_ID NUMBER, ORDER_DATE DATE, STORE STRING, AMOUNT NUMBER);
LIST @LANDING;

20日分と21日分のファイルも、日付・行数・注文番号の始まりを変えた同じ形の文で書き出しました。形式の違うファイルの書き出し方は、試す節で載せます。ORDERS の列は、自社の取り込み先のテーブルに合わせます。

記録テーブルには、1回の実行ごとに必ず1行を残す

記録テーブルは、ストアドプロシージャを1回呼ぶごとに1行を書くテーブルです。新しいファイルが無い回も、SQL が失敗した回も、1行を残します。

USE SCHEMA test_ingest.raw;
CREATE OR REPLACE TABLE LOAD_LOG (
  RUN_AT TIMESTAMP_LTZ,
  RESULT STRING,
  FILES_LOADED NUMBER,
  ROWS_LOADED NUMBER,
  FILES_SKIPPED NUMBER,
  MESSAGE STRING
);

RUN_AT は、ストアドプロシージャを始めた時刻です。RESULT は、その回の結果です。FILES_LOADED と ROWS_LOADED は、取り込んだファイルの数とデータの行数です。FILES_SKIPPED は、誤りがあって飛ばしたファイルの数です。

MESSAGE には、飛ばしたファイルのエラー文か、失敗した SQL のエラー文が入ります。飛ばしたファイルが2つ以上ある回は、そのうち1つのエラー文です。

RESULT に入る値は、OK、NO_FILE、PARTIAL、ERROR の4つです。値の名前は、当社が決めたものです。OK は、ファイルを1つ以上取り込み、飛ばしたファイルが無い回です。NO_FILE は、取り込んだファイルが0の回です。

PARTIAL は、飛ばしたファイルが1つ以上ある回です。ほかに新しいファイルがあれば、そちらは取り込みます。ERROR は、ストアドプロシージャの中の SQL が失敗した回です。

その日の行が1つも無いのは、ストアドプロシージャが呼ばれなかった日か、アラートが動く6時30分までに終わっていない日です。このとき、アラートは結果を NO_RUN と書きます。

タスクの実行履歴は過去7日分、取り込みの履歴は過去14日分までしか、INFORMATION_SCHEMA の関数では見られません。SNOWFLAKE.ACCOUNT_USAGE のビューなら365日分ありますが、反映が遅れます。記録テーブルは自分で作ったテーブルなので、削除しないかぎり残ります。

RESULT の値ごとに、担当が何を見て、どう直すかを先に決めておきます。次の表は、OK から ERROR までの結果を実際に出したうえで決めたものです。

ストアドプロシージャで COPY INTO を実行し、結果を記録テーブルに書く

ストアドプロシージャには、LOAD_ORDERS_DAILY という名前を付けました。中では、まず始めた時刻を控えて、COPY INTO を実行します。次に、取り込みの履歴を始めた時刻から見て、取り込んだファイルと飛ばしたファイルを数えます。最後に結果を決め、記録テーブルに1行書いて、結果を文字列で返します。

USE SCHEMA test_ingest.raw;
CREATE OR REPLACE PROCEDURE LOAD_ORDERS_DAILY()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  started_at TIMESTAMP_LTZ;
  files_loaded NUMBER DEFAULT 0;
  rows_loaded NUMBER DEFAULT 0;
  files_skipped NUMBER DEFAULT 0;
  first_error STRING;
  result STRING;
BEGIN
  started_at := CURRENT_TIMESTAMP();
  COPY INTO ORDERS FROM @LANDING/orders/
    FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1)
    ON_ERROR = SKIP_FILE;
  SELECT COALESCE(COUNT_IF(STATUS = 'Loaded'), 0),
         COALESCE(SUM(IFF(STATUS = 'Loaded', ROW_COUNT, 0)), 0),
         COALESCE(COUNT_IF(STATUS <> 'Loaded'), 0),
         MAX(FIRST_ERROR_MESSAGE)
    INTO :files_loaded, :rows_loaded, :files_skipped, :first_error
    FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(TABLE_NAME => 'ORDERS', START_TIME => :started_at));
  result := CASE WHEN files_skipped > 0 THEN 'PARTIAL'
                 WHEN files_loaded = 0 THEN 'NO_FILE'
                 ELSE 'OK' END;
  INSERT INTO LOAD_LOG VALUES (:started_at, :result, :files_loaded, :rows_loaded, :files_skipped, :first_error);
  RETURN result || ' files=' || files_loaded || ' rows=' || rows_loaded || ' skipped=' || files_skipped;
EXCEPTION
  WHEN OTHER THEN
    INSERT INTO LOAD_LOG VALUES (CURRENT_TIMESTAMP(), 'ERROR', 0, 0, 0, :SQLERRM);
    RAISE;
END;
$$;
CALL LOAD_ORDERS_DAILY();

取り込みの履歴は、COPY_HISTORY というテーブル関数で見ます。テーブル関数は、FROM 句に書くと結果をテーブルのように扱える関数です。テーブルの名前と数え始める時刻を指定すると、その時刻からあとに読んだファイルが1行ずつ出ます。

STATUS が Loaded のファイルは FILES_LOADED に、それ以外は FILES_SKIPPED に数えます。新しいファイルが無い回は、取り込みの履歴に何も出ないので、どちらも0になります。取り込みの履歴は COPY INTO の直後に見ていて、今回はどの回も数が合っていました。

見出しの行しか無いファイルが届いた日は、データが0行でも OK になるはずです。これも NO_FILE として扱うなら、CASE 式の2つ目の条件を書き換えます。書き換えたあとの条件は WHEN files_loaded = 0 OR rows_loaded = 0 THEN 'NO_FILE' です。見出しの行だけのファイルも、この書き換えも、今回は試していません。

SQL が失敗したときは、EXCEPTION の中で ERROR の行を書きます。そのあと、RAISE で同じ例外をもう一度発生させます。そうすると、ストアドプロシージャを呼んだタスクも失敗になり、タスクの実行履歴に FAILED が残ります。

COPY INTO には、ON_ERROR = SKIP_FILE を指定しました。誤りのあるファイルを、ファイルごと飛ばす指定です。ほかのファイルは取り込みます。CONTINUE は、誤りのある行だけを飛ばし、同じファイルの正しい行は取り込みます。

SKIP_FILE にしたのは、当社の判断です。CONTINUE にすると、1つのファイルのうち一部の行だけが入り、どの行が抜けたかをあとで探すことになります。ファイルごと飛ばしておけば、直したファイルを置き直すだけで済みます。なお、SKIP_FILE は誤りが無くてもファイル全体をいったん読み込むため、ほかの値より遅いとドキュメントにあります。

CONTINUE に変えるなら、数え方も見直します。この SQL は STATUS が Loaded 以外を飛ばした数に入れるので、一部だけ入ったファイルも飛ばしたファイルに数えます。

上の SQL は、一度直したあとのものです。最初に書いた版では、新しいファイルが無い回が OK と記録されました。次の画面の1行目です。

この回は、取り込みの履歴に行が1つもありませんでした。そのとき COUNT_IF の結果が、0ではなく NULL になりました。NULL は0と比べても真にならないので、CASE 式は最後の ELSE に進み、OK を返しました。

直したのは、COUNT_IF を COALESCE で包み、NULL のときに0にすることです。ストアドプロシージャを作ったら、新しいファイルが無い状態で一度呼び、記録テーブルに NO_FILE と書かれるかを確かめてください。

新しいファイルが無い回と、形式の違うファイルが混ざった回を試す

直したストアドプロシージャで、土台を作り直して最初から試しました。2026年10月21日の0時台に、19日分、20日分、21日分の注文のファイルを続けて置いています。

19日分のファイルの100行を取り込んだ回は、OK です。次に、ファイルを足さずにもう一度呼ぶと、NO_FILE と記録されました。

続けて、20日分のファイルを2つ置きました。1つは、いつもの形式の orders_2026-10-20.csv で、120行あります。もう1つは orders_2026-10-20_shopB.csv で、1行だけです。2つ目は、別の店から届いたことにして作ったファイルで、注文番号に英字を、金額に桁区切りのカンマを入れてあります。次の文で書き出しました。

USE SCHEMA test_ingest.raw;
COPY INTO @LANDING/orders/orders_2026-10-20_shopB.csv
  FROM (SELECT 'A-0001' AS ORDER_ID, '2026-10-20' AS ORDER_DATE, '池袋店' AS STORE, '3,200' AS AMOUNT)
  FILE_FORMAT = (TYPE = CSV COMPRESSION = NONE) HEADER = TRUE SINGLE = TRUE OVERWRITE = TRUE;

ストアドプロシージャを呼ぶと、結果は PARTIAL でした。取り込んだのは1つのファイルの120行で、飛ばしたファイルは1つです。取り込み先のテーブルの件数は220になりました。

形式の違うファイルを置いたまま、もう一度呼ぶと、また PARTIAL になりました。外すか直すまで、呼ぶたびに PARTIAL が続きます。

写真の ORDERS_ROWS は、記録テーブルを見る SELECT の中で、取り込み先のテーブルを数えた値です。SELECT を実行した時点の件数なので、どの行にも同じ値が出ます。

取り込みの履歴も、数え始める時刻を2時間前にして見ました。

USE SCHEMA test_ingest.raw;
SELECT FILE_NAME, STATUS, ROW_COUNT, ERROR_COUNT, FIRST_ERROR_MESSAGE
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(TABLE_NAME => 'ORDERS', START_TIME => DATEADD(HOUR, -2, CURRENT_TIMESTAMP())))
ORDER BY LAST_LOAD_TIME DESC;

飛ばしたファイルは、STATUS が Load failed の行として、呼んだ回ごとに残りました。NO_FILE の回は、行が出ていません。メールが届いた日にこの SQL を使うときは、START_TIME を2時間前ではなく、その日のタスクの時刻からにします。

FIRST_ERROR_MESSAGE には、Numeric value 'A-0001' is not recognized と出ました。金額のカンマの誤りは、エラー文に出ていません。エラー文に出るのは、最初の誤り1つだけでした。ERROR_COUNT は1で、誤りのあるデータの行数です。

注文番号だけを直しても、金額の誤りが残っていれば、そのファイルはまた飛ばされるはずです。送り手には、ファイル全体を確かめてもらいます。

タスクにして、予定の時刻に動くかを確かめる

タスクには、ストアドプロシージャを呼ぶ CALL を1文だけ書きます。

USE SCHEMA test_ingest.raw;
CREATE OR REPLACE TASK LOAD_ORDERS
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 0 6 * * * Asia/Tokyo'
  AS CALL LOAD_ORDERS_DAILY();

SCHEDULE には、cron の式とタイムゾーンを続けて書きます。cron の式は、分・時・日・月・曜日の順の5つの欄です。USING CRON 0 6 * * * Asia/Tokyo は、日本時間の毎朝6時です。

WAREHOUSE を書くと、そのウェアハウスで動き、起動するたびに最低1分ぶんの料金がかかります。WAREHOUSE を省くとサーバーレスタスクになり、大きさは Snowflake が自動で決めます。初めの大きさを指定しないと、MEDIUM のウェアハウスにあたる大きさで始まります。

初めの大きさは、USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE で指定します。料金のかかり方は、料金とコストの記事にまとめています。

予定の時刻で動かす前に、SQL が失敗したときの動きを確かめました。形式の違うファイルを REMOVE で外し、取り込み先のテーブルの名前を変えてから、EXECUTE TASK でタスクを手動で1回動かしました。

記録テーブルには、ERROR の行が入りました。MESSAGE は SQL compilation error: Table 'ORDERS' does not exist です。タスクの実行履歴は FAILED で、ERROR_CODE は1757でした。

タスクの実行履歴の ERROR_MESSAGE は、次の文で始まっていました。

Uncaught exception of type 'STATEMENT_ERROR' on line 11 …

記録テーブルとタスクの実行履歴の、どちらにも失敗が残りました。写真のように、タスクの名前で実行履歴を見ると前の実行も出るので、その日の分は SCHEDULED_TIME_RANGE_START で絞ります。

テーブルの名前を元に戻し、21日分のファイル orders_2026-10-21.csv を置きました。80行のファイルです。作ったばかりのタスクは、一時停止の状態から始まります。予定の時刻に動かすには、ALTER TASK … RESUME を実行します。

今回は、試すために時刻を数分先の0:31 に変えてから RESUME しました。

USE SCHEMA test_ingest.raw;
ALTER TASK LOAD_ORDERS SET SCHEDULE = 'USING CRON 31 0 * * * Asia/Tokyo';
ALTER TASK LOAD_ORDERS RESUME;
SHOW TASKS LIKE 'LOAD_ORDERS' IN SCHEMA test_ingest.raw;

0:31 を過ぎてから、タスクの実行履歴を直近1時間に絞って見ました。そのあと、タスクを一時停止に戻しました。

USE SCHEMA test_ingest.raw;
SELECT NAME, STATE, ERROR_CODE,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'YYYY-MM-DD HH24:MI:SS') AS SCHEDULED_JST,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', COMPLETED_TIME), 'YYYY-MM-DD HH24:MI:SS') AS COMPLETED_JST
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(TASK_NAME => 'LOAD_ORDERS',
       SCHEDULED_TIME_RANGE_START => DATEADD(HOUR, -1, CURRENT_TIMESTAMP())))
ORDER BY SCHEDULED_TIME DESC;
ALTER TASK LOAD_ORDERS SUSPEND;

タスクは0:31 に予定どおり動き、3秒で終わりました。記録テーブルには OK の行が入り、取り込んだのは1つのファイルの80行です。取り込み先のテーブルの件数は300で、100＋120＋80 と合います。ここまでの記録テーブルの行を並べると、次のようになります。

タスクは、続けて10回失敗すると、自動で一時停止します。SUSPEND_TASK_AFTER_NUM_FAILURES の既定が10だからです。

アラートが記録テーブルを毎朝見て、問題がある日だけメールで知らせる

アラートは、決めた時刻に条件の SQL を実行し、結果が1行以上あれば、決めておいた処理を実行する仕組みです。今回は、タスクの30分後の毎朝6時30分に動かします。条件は、その日の記録テーブルに行が無いか、最後の行の RESULT が OK でないことです。

その日かどうかは、日本時間の日付で判定しています。記録テーブルを人が見るだけなら、担当は毎朝 SELECT を実行し、結果を読みに行く必要があります。アラートがメールを送る形にすれば、問題がある日だけ知らせが届くので、毎朝見に行かなくてもよくなります。

先に、メールを送らない形でも試しました。処理を、LOAD_ALERTS というテーブルに1行書くだけにしたアラートです。この SQL は載せていません。その日の最後の行が0:31 の OK の間は、アラートの実行履歴の STATE は CONDITION_FALSE でした。ファイルを足さずにストアドプロシージャを呼んで NO_FILE の行を足し、もう一度アラートを実行すると、TRIGGERED になりました。

メールの宛先にできるのは、同じアカウントの Snowflake ユーザーで、メールアドレスを確認済みの人だけです。試用の環境のユーザーは1人だけで、そのアドレスは宛先にしたいものではありませんでした。そこで、受け取り用のユーザーをパスワード無しで作りました。

USE ROLE ACCOUNTADMIN;
USE SCHEMA test_ingest.raw;
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');

SYSTEM$START_USER_EMAIL_VERIFICATION を呼ぶと、そのアドレスに確認メールが送られます。受け取った人が確認メールのリンクを開くと、アドレスは確認済みになります。今回も、受け取る人にリンクを開いてもらってから先へ進みました。

この節の SQL では、宛先のアドレスとユーザー名を、説明のために作った名前に置き換えています。通知インテグレーションの名前と、ユーザーのコメントも同じです。data-team@example.com は、自社の確認済みのアドレスにして使います。

通知インテグレーションは、Snowflake からメールなどで外へ知らせるための設定です。ALLOWED_RECIPIENTS には、送ってよい宛先を書きます。知らせる処理は、ストアドプロシージャ NOTIFY_LOAD_RESULT にまとめました。その日の最後の結果とエラー文を LOAD_ALERTS に書き、SYSTEM$SEND_EMAIL でメールを送ります。

当社は、0時台の試しを片付けたあと、1時台に試す土台を作り直してメールを試しました。続けて試すときは、片付けずにそのまま次の塊を実行してかまいません。

次の塊は、記録テーブルと LOAD_ALERTS を作り直すので、それまでの記録テーブルの行は消えます。そのうえで、試すための PARTIAL の行を1つ INSERT で入れ、アラートを手動で動かします。この行に対応する取り込みは、していません。

USE ROLE ACCOUNTADMIN;
USE SCHEMA test_ingest.raw;
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS LOAD_EMAIL_INT
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('data-team@example.com');
-- 記録テーブルと LOAD_ALERTS を作り直す（それまでの行は消える）
CREATE OR REPLACE TABLE LOAD_LOG (RUN_AT TIMESTAMP_LTZ, RESULT STRING, FILES_LOADED NUMBER, ROWS_LOADED NUMBER, FILES_SKIPPED NUMBER, MESSAGE STRING);
CREATE OR REPLACE TABLE LOAD_ALERTS (DETECTED_AT TIMESTAMP_LTZ, LAST_RESULT STRING, MESSAGE STRING);
-- 試すための行（取り込みはしていない）。本番では入れない
INSERT INTO LOAD_LOG VALUES (CURRENT_TIMESTAMP(), 'PARTIAL', 1, 120, 1, 'Numeric value ''A-0001'' is not recognized');
CREATE OR REPLACE PROCEDURE NOTIFY_LOAD_RESULT()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  last_result STRING;
  last_message STRING;
  body STRING;
BEGIN
  SELECT COALESCE(MAX_BY(RESULT, RUN_AT), 'NO_RUN'), MAX_BY(MESSAGE, RUN_AT)
    INTO :last_result, :last_message
    FROM LOAD_LOG
    WHERE CONVERT_TIMEZONE('Asia/Tokyo', RUN_AT)::DATE = CONVERT_TIMEZONE('Asia/Tokyo', CURRENT_TIMESTAMP())::DATE;
  INSERT INTO LOAD_ALERTS VALUES (CURRENT_TIMESTAMP(), :last_result, :last_message);
  body := 'これはブログ記事の検証用のテスト送信です。' || CHR(10) ||
          '本日の取り込みの結果: ' || last_result || CHR(10) ||
          'メッセージ: ' || COALESCE(last_message, '(なし)') || CHR(10) ||
          '記録テーブル LOAD_LOG の本日の行を確認してください。';
  CALL SYSTEM$SEND_EMAIL(
    'LOAD_EMAIL_INT',
    'data-team@example.com',
    '[テスト送信] Snowflake 取り込みの確認（ブログ検証）',
    :body);
  RETURN last_result;
END;
$$;
CREATE OR REPLACE ALERT CHECK_DAILY_LOAD_MAIL
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 30 6 * * * Asia/Tokyo'
  IF (EXISTS (
    SELECT 1
    FROM (SELECT COUNT(*) AS RUNS, MAX_BY(RESULT, RUN_AT) AS LAST_RESULT
          FROM LOAD_LOG
          WHERE CONVERT_TIMEZONE('Asia/Tokyo', RUN_AT)::DATE = CONVERT_TIMEZONE('Asia/Tokyo', CURRENT_TIMESTAMP())::DATE)
    WHERE RUNS = 0 OR LAST_RESULT <> 'OK'
  ))
  THEN
    CALL NOTIFY_LOAD_RESULT();
EXECUTE ALERT CHECK_DAILY_LOAD_MAIL;

件名と本文は、試したときのままです。アラートも作った直後は一時停止の状態なので、EXECUTE ALERT で手動で1回動かしました。

動かしたあとは、アラートの実行履歴と LOAD_ALERTS を、次の SELECT で1つの結果に並べて見ました。

USE SCHEMA test_ingest.raw;
SELECT 'ALERT' AS SRC, STATE AS VALUE1, COALESCE(SQL_ERROR_MESSAGE, 'null') AS VALUE2,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'YYYY-MM-DD HH24:MI:SS') AS AT_JST
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(SCHEDULED_TIME_RANGE_START => DATEADD(HOUR, -1, CURRENT_TIMESTAMP())))
WHERE NAME = 'CHECK_DAILY_LOAD_MAIL'
UNION ALL
SELECT 'LOAD_ALERTS', LAST_RESULT, MESSAGE, TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', DETECTED_AT), 'YYYY-MM-DD HH24:MI:SS')
FROM LOAD_ALERTS
ORDER BY AT_JST DESC;

アラートの実行履歴は TRIGGERED で、LOAD_ALERTS には PARTIAL とエラー文の行が入りました。TRIGGERED は、条件が成り立ち、処理も成功したという意味です。メールを送ったかどうかは、NOTIFICATION_HISTORY というテーブル関数で見ます。

USE SCHEMA test_ingest.raw;
SELECT INTEGRATION_NAME, STATUS, COALESCE(ERROR_MESSAGE, 'null') AS ERROR_MESSAGE,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', CREATED), 'YYYY-MM-DD HH24:MI:SS') AS CREATED_JST,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', PROCESSED), 'YYYY-MM-DD HH24:MI:SS') AS PROCESSED_JST
FROM TABLE(INFORMATION_SCHEMA.NOTIFICATION_HISTORY(START_TIME => DATEADD(HOUR, -1, CURRENT_TIMESTAMP())))
ORDER BY CREATED DESC;

STATUS は SUCCESS で、ERROR_MESSAGE は空でした。SUCCESS は、Snowflake がメールを送り出したという記録で、受信箱に届いたことまでは示しません。今回は、宛先の受信箱にも届き、日本語の本文も読めることを、受け取った人に確かめてもらいました。本番で動かし始めたときも、最初の1通が届いたかを、受け取る人に受信箱で確かめてもらいます。

このアラートは、その日の最後の行だけで判定します。PARTIAL のあとに誰かが手で呼び直して OK になれば、メールは届きません。

アラートそのものが失敗した日も、メールは届きません。アラートの実行履歴に CONDITION_FAILED か ACTION_FAILED が出たら、アラートの SQL が失敗しています。アラートは、条件や処理の SQL に誤った名前があっても、作るときにはエラーになりません。作ったら一度 EXECUTE ALERT で動かして確かめます。

タスクには、失敗を外へ知らせる ERROR_INTEGRATION という設定もありますが、これではメールは送れません。Amazon SNS などに送るための設定です。アラートもウェアハウスで動かすと、起動のたびに最低1分ぶんの料金がかかります。WAREHOUSE を省けば、サーバーレスのアラートになります。

メールが届いた日は、次の2つの SELECT で、記録テーブルと LOAD_ALERTS を新しい順に見ます。記録テーブルの行が増えてきたら、WHERE で RUN_AT を直近の数日に絞ります。

USE SCHEMA test_ingest.raw;
SELECT TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', RUN_AT), 'YYYY-MM-DD HH24:MI:SS') AS RUN_AT_JST,
       RESULT, FILES_LOADED, ROWS_LOADED, FILES_SKIPPED,
       (SELECT COUNT(*) FROM ORDERS) AS ORDERS_ROWS, MESSAGE
FROM LOAD_LOG ORDER BY RUN_AT DESC;
SELECT TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', DETECTED_AT), 'YYYY-MM-DD HH24:MI:SS') AS DETECTED_JST, LAST_RESULT, MESSAGE
FROM LOAD_ALERTS ORDER BY DETECTED_AT DESC;

PARTIAL と NO_FILE の日に使う SQL と、ファイルの置き直し方

PARTIAL の日は、まず取り込みの履歴の Load failed の行で、飛ばしたファイルの名前を見ます。

直すときは、形式の違うファイルを REMOVE でステージから外してから、直したファイルを置きます。置いたままだと、呼ぶたびに読まれて PARTIAL が続くからです。置いたら、ストアドプロシージャを呼び直します。

USE SCHEMA test_ingest.raw;
LIST @LANDING;
REMOVE @LANDING/orders/orders_2026-10-20_shopB.csv;
CALL LOAD_ORDERS_DAILY();

内部ステージにファイルを置く SQL は PUT です。PUT は、Snowsight で SQL を書く画面であるワークシートからは実行できません。Snowflake CLI やドライバから実行します。Snowsight の画面から、名前付きの内部ステージへアップロードすることもできます。

PUT は、既定でファイルを gzip に圧縮します。同じ名前のファイルがステージにあると、既定では上書きしません。直したファイルを置いて呼び直す手順は、今回は試していません。

中身を変えて置き直したファイルは、同じ名前でも新しいファイルとして取り込まれます。前に取り込んだファイルと同じ名前・同じ中身のファイルは、呼び直しても読み飛ばされます。ロードのメタデータは64日で消えます。64日を過ぎて取り込んだかどうか分からなくなったファイルも、既定では読み飛ばされます。

ただし FORCE = TRUE を指定すると、取り込み済みのファイルも読み直すので、同じデータが二重に入るおそれがあります。二重に入ったテーブルを、取り込む前の状態に戻す方法は、Time Travel と UNDROP の記事に書きました。

NO_FILE の日は、LIST でステージにその日のファイルがあるかを見ます。無ければ、送り手に置いてもらってから、ストアドプロシージャを呼びます。LIST に出ているのに NO_FILE なら、orders/ の外に置いたか、取り込み済みのファイルを置き直したことが考えられます。CASE 式を書き換えた場合は、見出しの行だけのファイルが届いた日も NO_FILE になります。

本番に入れる SQL

本番に入れる一式は、ここにまとめました。作るのは、記録テーブル、LOAD_ALERTS、ストアドプロシージャ2つ、タスク、通知インテグレーション、メールを送るアラートです。最後に RESUME します。試すための INSERT・CALL・EXECUTE ALERT・REMOVE、メールを送らないアラート、0:31 の時刻は入れていません。

<…> の所を、自社の名前・宛先・件名・本文・時刻に書き換えます。書き換える所は次のとおりです。

• データベースとスキーマ、ウェアハウス：<データベース>.<スキーマ> と <ウェアハウス>。
• ファイルの置き場所と取り込み先：<ステージ>/<パス>/ と <取り込み先のテーブル>。テーブルの名前は、COPY_HISTORY に渡す TABLE_NAME にも入れます。
• ファイルの形式：FILE_FORMAT を自社の CSV に合わせます。今回は見出しの行が1行あるので、SKIP_HEADER = 1 です。
• メール：<通知インテグレーション>、<宛先>、<件名>、<本文の書き出し>。宛先は、同じアカウントの確認済みのユーザーのアドレスです。
• 時刻：タスクの <タスクの分> <タスクの時> はファイルが届いたあと、アラートの時刻はタスクが終わったあとにします。
• タイムゾーン：<タイムゾーン> は、タスクとアラートの時刻と、その日を判定する日付の両方に使います。

記録テーブルと LOAD_ALERTS の CREATE OR REPLACE TABLE は、最初の1回だけ実行します。2回目以降に実行すると、それまでの記録が消えます。また、取り込み先のテーブルを作り直すと、そのテーブルの取り込みの履歴は消え、置いたままのファイルをまた取り込むはずです。ロードのメタデータもテーブルごとに持つからです。

-- 実行した文から試すための文を除き、名前を置き換えた形。この形のまとまりでは実行していない
-- <…> は、自社の名前・宛先・件名・本文・時刻に書き換える
USE ROLE ACCOUNTADMIN;
USE WAREHOUSE <ウェアハウス>;
USE SCHEMA <データベース>.<スキーマ>;
-- 記録テーブルと LOAD_ALERTS（最初の1回だけ実行する）
CREATE OR REPLACE TABLE LOAD_LOG (
  RUN_AT TIMESTAMP_LTZ,
  RESULT STRING,
  FILES_LOADED NUMBER,
  ROWS_LOADED NUMBER,
  FILES_SKIPPED NUMBER,
  MESSAGE STRING
);
CREATE OR REPLACE TABLE LOAD_ALERTS (DETECTED_AT TIMESTAMP_LTZ, LAST_RESULT STRING, MESSAGE STRING);
-- 取り込みのストアドプロシージャ
CREATE OR REPLACE PROCEDURE LOAD_ORDERS_DAILY()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  started_at TIMESTAMP_LTZ;
  files_loaded NUMBER DEFAULT 0;
  rows_loaded NUMBER DEFAULT 0;
  files_skipped NUMBER DEFAULT 0;
  first_error STRING;
  result STRING;
BEGIN
  started_at := CURRENT_TIMESTAMP();
  COPY INTO <取り込み先のテーブル> FROM @<ステージ>/<パス>/
    FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1)
    ON_ERROR = SKIP_FILE;
  SELECT COALESCE(COUNT_IF(STATUS = 'Loaded'), 0),
         COALESCE(SUM(IFF(STATUS = 'Loaded', ROW_COUNT, 0)), 0),
         COALESCE(COUNT_IF(STATUS <> 'Loaded'), 0),
         MAX(FIRST_ERROR_MESSAGE)
    INTO :files_loaded, :rows_loaded, :files_skipped, :first_error
    FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(TABLE_NAME => '<取り込み先のテーブル>', START_TIME => :started_at));
  result := CASE WHEN files_skipped > 0 THEN 'PARTIAL'
                 WHEN files_loaded = 0 THEN 'NO_FILE'
                 ELSE 'OK' END;
  INSERT INTO LOAD_LOG VALUES (:started_at, :result, :files_loaded, :rows_loaded, :files_skipped, :first_error);
  RETURN result || ' files=' || files_loaded || ' rows=' || rows_loaded || ' skipped=' || files_skipped;
EXCEPTION
  WHEN OTHER THEN
    INSERT INTO LOAD_LOG VALUES (CURRENT_TIMESTAMP(), 'ERROR', 0, 0, 0, :SQLERRM);
    RAISE;
END;
$$;
-- タスク
CREATE OR REPLACE TASK LOAD_ORDERS
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <タスクの分> <タスクの時> * * * <タイムゾーン>'
  AS CALL LOAD_ORDERS_DAILY();
-- メールの通知インテグレーションと、知らせるストアドプロシージャ
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS <通知インテグレーション>
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('<宛先>');
CREATE OR REPLACE PROCEDURE NOTIFY_LOAD_RESULT()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  last_result STRING;
  last_message STRING;
  body STRING;
BEGIN
  SELECT COALESCE(MAX_BY(RESULT, RUN_AT), 'NO_RUN'), MAX_BY(MESSAGE, RUN_AT)
    INTO :last_result, :last_message
    FROM LOAD_LOG
    WHERE CONVERT_TIMEZONE('<タイムゾーン>', RUN_AT)::DATE = CONVERT_TIMEZONE('<タイムゾーン>', CURRENT_TIMESTAMP())::DATE;
  INSERT INTO LOAD_ALERTS VALUES (CURRENT_TIMESTAMP(), :last_result, :last_message);
  body := '<本文の書き出し>' || CHR(10) ||
          '本日の取り込みの結果: ' || last_result || CHR(10) ||
          'メッセージ: ' || COALESCE(last_message, '(なし)') || CHR(10) ||
          '記録テーブル LOAD_LOG の本日の行を確認してください。';
  CALL SYSTEM$SEND_EMAIL(
    '<通知インテグレーション>',
    '<宛先>',
    '<件名>',
    :body);
  RETURN last_result;
END;
$$;
-- メールを送るアラート
CREATE OR REPLACE ALERT CHECK_DAILY_LOAD_MAIL
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <アラートの分> <アラートの時> * * * <タイムゾーン>'
  IF (EXISTS (
    SELECT 1
    FROM (SELECT COUNT(*) AS RUNS, MAX_BY(RESULT, RUN_AT) AS LAST_RESULT
          FROM LOAD_LOG
          WHERE CONVERT_TIMEZONE('<タイムゾーン>', RUN_AT)::DATE = CONVERT_TIMEZONE('<タイムゾーン>', CURRENT_TIMESTAMP())::DATE)
    WHERE RUNS = 0 OR LAST_RESULT <> 'OK'
  ))
  THEN
    CALL NOTIFY_LOAD_RESULT();
-- 動かし始める（ALTER ALERT … RESUME と SHOW ALERTS は、今回は実行していない）
ALTER TASK LOAD_ORDERS RESUME;
ALTER ALERT CHECK_DAILY_LOAD_MAIL RESUME;
SHOW TASKS LIKE 'LOAD_ORDERS' IN SCHEMA <データベース>.<スキーマ>;
SHOW ALERTS LIKE 'CHECK_DAILY_LOAD_MAIL' IN SCHEMA <データベース>.<スキーマ>;

通知インテグレーションは、試したときとは別の名前で作ります。同じ名前のものが残っていると、IF NOT EXISTS のため作り直されません。ALLOWED_RECIPIENTS も、試したときの宛先のままになります。別の名前にしておけば、片付けで試したときの通知インテグレーションを削除しても、本番のメールには影響しません。

あとでタスクの時刻を変えるときは、先に ALTER TASK … SUSPEND で一時停止します。単独のタスクは、一時停止していないと変更できないからです。変えたあとは、RESUME し直します。

ロールと権限は、今回は ACCOUNTADMIN で試しました。ほかのロールで作るなら、次の権限が要ります。

• タスク：データベースの USAGE、スキーマの USAGE と CREATE TASK、ウェアハウスの USAGE。RESUME のときには、タスクを所有するロールがアカウントの EXECUTE TASK を持っているかが確かめられ、無ければエラーになります。サーバーレスなら EXECUTE MANAGED TASK も要ります。
• ストアドプロシージャ：スキーマの CREATE PROCEDURE。ほかのロールに呼ばせるなら、そのストアドプロシージャの USAGE を付与します。
• ストアドプロシージャの中の SQL：ストアドプロシージャは既定で所有するロールの権限で動きます。そのロールには、取り込み先のテーブルと記録テーブルに書く権限、ステージのファイルを読む権限などが要ります。
• アラート：アカウントの EXECUTE ALERT。付与できるのは ACCOUNTADMIN のロールだけです。ほかに、スキーマの USAGE と CREATE ALERT、データベースとウェアハウスの USAGE が要ります。条件で LOAD_LOG を読むので、その権限も要ります。
• メール：通知インテグレーションを作るには、アカウントの権限が要ります。CREATE INTEGRATION か CREATE NOTIFICATION INTEGRATION です。SYSTEM$SEND_EMAIL を呼ぶのは NOTIFY_LOAD_RESULT です。それを所有するロールに、通知インテグレーションの USAGE を付与します。

ロールに権限を付与して確かめる進め方は、行アクセスポリシーの記事に書きました。

試し終えたら、検証用の test_ingest と受け取り用のユーザーを片付ける

試し終えたら、タスクとアラートを一時停止してから、test_ingest を削除します。データ保持期間を0にしておくと、削除したあとに Time Travel で戻せる期間も残りません。受け取り用のユーザーと通知インテグレーションはアカウントの物なので、データベースを削除しても残ります。試すためだけに作ったなら、これも削除します。

次の塊の各行は、実行した文です。ただし、この形のまとまりでは実行していません。SUSPEND と DROP DATABASE は試しのたびに、DROP USER と DROP INTEGRATION は検証の最後に実行しました。

USE ROLE ACCOUNTADMIN;
USE SCHEMA test_ingest.raw;
ALTER TASK test_ingest.raw.LOAD_ORDERS SUSPEND;
ALTER ALERT test_ingest.raw.CHECK_DAILY_LOAD_MAIL SUSPEND;
ALTER DATABASE test_ingest SET DATA_RETENTION_TIME_IN_DAYS = 0;
DROP DATABASE test_ingest;
-- 受け取り用のユーザーと通知インテグレーション（検証の最後に実行した）
DROP USER IF EXISTS DATA_TEAM_NOTIFY;
DROP INTEGRATION IF EXISTS LOAD_EMAIL_INT;

動かし始めた最初の数日は、タスク・記録テーブル・アラートの3つを並べて見る

本番で動かし始めた最初の数日は、メールを待つだけにせず、次の3つを毎朝並べて見ます。1つ目は、予定の時刻の回がタスクの実行履歴にあることです。2つ目は、同じ時刻の行が記録テーブルにあることです。3つ目は、そのあとアラートの実行履歴の行が増えていることです。

この3つを見るのは、今回確かめていないことがあるからです。アラートを RESUME して6時30分に動くところと、タスクが毎日続けて動くところは見ていません。取り込みの履歴の反映が遅れて、数え漏れることがあるかも確かめていません。

サーバーレスのタスクとアラート、Snowpipe、外部ステージ、ON_ERROR = CONTINUE も試していません。取り込み先のテーブルを毎朝作り直す運用、権限を絞ったロールでの運用、今回の料金も見ていません。今回の結果は、試用の Snowflake の1つのアカウントで続けて試したものです。Enterprise Edition・AWS の東京で、ACCOUNTADMIN と X-Small のウェアハウスを使いました。

この一式を、自社のファイルの置き方や取り込む頻度に合わせて組むところは、当社でもお手伝いできます。Snowflake 導入支援のうち、データ取り込みの構築の工程にあたります。