Snowflake の Stream は読むだけでは消費できないので Task に消費させる

こんにちは、データ事業本部のまっきーです。

業務で Snowflake 上に文書処理パイプラインを構築しています。ステージに着地したファイルを AI 関数でパースし、チャンク化やメタデータ抽出などの加工を経て検索可能にします。この一連の流れを Stream とサーバーレス Task の DAG で全自動化しました。

正直、Stream と Task は「ドキュメントを読めば分かるだろう」と思って触り始めました。実際に組んでみると設計の考えどころが想像より多く、自分の整理も兼ねて残しておきたくなった、というのがこの記事の動機です。

実運用で使うには「Stream の消費」「失効」「Task の作り直し」の理解が必要で、設計時に調べて回避したポイントが 7 つ溜まりました。

本記事はその 7 つを、実際に組んだ構成のコード付きでまとめたものです。

Stream は、テーブルへの変更（INSERT / UPDATE / DELETE）を記録しておいてくれる Snowflake の変更追跡オブジェクトです。Task は SQL を定期実行するスケジューラで、AFTER で依存関係を組むと DAG（有向グラフ）として連結できます。

作ったパイプラインの全体像

題材は汎用的な文書処理パイプラインです。ファイルのメタデータが台帳テーブルに登録されると、それを起点に下流の処理が順に走ります。

時系列で追うとこうなります。

1. 新しいファイルがステージに着地し、上流処理が台帳テーブルにメタデータを 1 行 INSERT する
2. その瞬間、Stream が台帳テーブルの変更を自動で記録する
3. 最大 10 分後にルート Task が起床し、Stream に未消費の差分があるかを判定する
4. 差分があれば DAG が起動し、子 Task たちが未処理のファイルを処理する（なければその回は何もしない）

Stream の役割は「DAG を起動するかどうかの合図」までです。どのファイルを処理するかは Stream の中身ではなく、各 Task が処理対象テーブルの未処理状態を見て決めます（この差分処理の設計は後半のポイントで繰り返し効いてきます）。

ルート Task の構えはこうです。

CREATE TABLE IF NOT EXISTS PIPELINE_RUN_LOG (
    RUN_AT TIMESTAMP_NTZ,   -- 起動時刻
    PENDING_CHANGES NUMBER  -- 起動時点の Stream 内の変更行数
);  -- 実行ログテーブル。ルート Task が Stream を消費するための DML 先を兼ねる

CREATE STREAM IF NOT EXISTS FILE_CHANGES_STREAM
ON TABLE RAW_FILES;  -- 台帳テーブルの変更を捕捉

CREATE TASK IF NOT EXISTS PIPELINE_ROOT_TASK
    SCHEDULE = '10 MINUTE'
    USER_TASK_TIMEOUT_MS = 1800000
    TASK_AUTO_RETRY_ATTEMPTS = 2
WHEN SYSTEM$STREAM_HAS_DATA('MYDB.MYSCHEMA.FILE_CHANGES_STREAM')
AS
    -- この INSERT が Stream を「消費」する（ポイント 1 で後述）
    INSERT INTO MYDB.MYSCHEMA.PIPELINE_RUN_LOG (RUN_AT, PENDING_CHANGES)
    SELECT CURRENT_TIMESTAMP(), COUNT(*)
    FROM MYDB.MYSCHEMA.FILE_CHANGES_STREAM;

SCHEDULE で 10 分ごとに目を覚まし、WHEN 句が起動ゲートになります。SYSTEM$STREAM_HAS_DATA(...) は「その Stream に未消費の差分があるか」を返すシステム関数で、差分がなければ本体の INSERT は実行されません。WAREHOUSE 句を書かないとサーバーレス Task（ウェアハウスを自分で指定せず、Snowflake 管理のコンピュートで動く実行方式）になります。以降の子 Task は AFTER PIPELINE_ROOT_TASK でぶら下げていくだけです。

7 つのポイント

1. Stream は読むだけでは消費できない

一番の落とし穴がこれでした。まず用語の整理からです。Stream は「差分をどこまで読んだか」を指すオフセット（しおり）を持っていて、これが進むことを「消費」と呼びます。消費されて初めて、その差分は処理済みの扱いになります。

重要なのは、オフセットの進む条件が 1 つしかないことです。進むのは、その Stream を INSERT や MERGE など DML 文の中で読んだときだけ。WHEN SYSTEM$STREAM_HAS_DATA(...) は「差分があるか」を覗くだけの起動判定なので進みませんし、素の SELECT で中身を眺めても進みません。「読むだけでは消費できない」とはこのことです。

ルート Task の本体が Stream を読まない SQL だと、次のことが起きます。

1. 台帳に変更が来て、Stream に差分が溜まる
2. 10 分後に WHEN が真になり Task が起動する。しかし本体は Stream を消費しない
3. 差分は残ったままなので、10 分後にまた WHEN が真になる
4. 以降、新しい変更が来なくても DAG 全体が 10 分ごとに起動し続ける

ルート Task の本体を「実行ログテーブルへの INSERT」にしているのはこのためです。INSERT ... SELECT ... FROM <stream> は DML の中で Stream を読む形なので、実行のたびにオフセットが進みます。起動記録を残す実益と、消費の仕掛けの一石二鳥です。

2. WHEN の起動判定はクラウドサービス課金

「10 分ごとに判定が走るとお金がかかるのでは」と最初は思いましたが、WHEN の条件評価はウェアハウスを起動せず、クラウドサービスレイヤーで実行されます。認証やメタデータ管理を担っている Snowflake の管理層のことで、ウェアハウスとは課金の仕組みが別です。条件が偽で本体が動かなかった回にも僅少な使用量は積まれますが、クラウドサービスの使用量は 1 日（UTC）の仮想ウェアハウス使用量の 10% を超えた分だけ課金される仕組みです。

1 点だけ注意があります。この 10% の分母は仮想ウェアハウスの使用量で、サーバーレス Task のコンピュートは含まれません。今回のようにサーバーレス Task 中心の構成では、この枠に収まるかはアカウント全体のウェアハウス使用量次第です。私の環境ではウェアハウスの通常利用があるため枠内に収まり、判定だけが走る待機時間のコストは実質かかっていませんでした。

条件が偽ならその回は SKIPPED になり、タスク履歴にもそのように残ります。SKIPPED は異常でなく「差分がなかった」という正常な記録です。

3. 自動リトライは冪等性とセット

TASK_AUTO_RETRY_ATTEMPTS = 2 をルート Task に付けると、DAG のどこかで失敗したときに最大 2 回まで自動で再実行してくれます。AI 関数のスロットリングのような一過性のエラー対策として便利です。

ただしこれを安全に使うには、各 Task の処理が冪等（何度実行しても結果が同じ）であることが前提になります。私のパイプラインでは全処理を「未処理・変更分だけを対象にする差分処理」として書いているため、リトライで同じ処理が再実行されても二重処理や AI の再課金は起きません。

逆に冪等でないとどうなるか。「処理済みかどうかを見ずに全ファイルをパースする」Task が 100 件中 90 件目の一過性エラーで失敗したとします。リトライは全件を投げ直すので、処理済みの 90 件が再パースされて AI の課金が積み増しされ、結果テーブルには同じファイルの結果が重複して入ります。重複はそのまま下流のチャンク化・検索まで流れ、しかも最後のリトライが成功すればタスク履歴は SUCCEEDED で終わり、事故が正常終了の顔で残ります。洗い替え型（毎回全件 DELETE → INSERT）の処理も同じ理屈でリトライとの相性が悪いです。リトライ設定は「付ければ安心」ではなく、冪等性の裏付けとセットで入れるものだと考えています。

4. Stream は止めている間に失効する

ポイント 1 で出てきたオフセット（しおり）には続きがあります。Stream は変更データのコピーを持っておらず、しおりが指しているのは、ソーステーブルが内部に持つ変更履歴です。この履歴はデータ保持期間を過ぎると消えていくため、消費されないまま放置すると、しおりが指す先の履歴のほうが先に消えてしまいます。こうして差分を計算できなくなった状態が失効（Snowflake の用語では stale。SHOW STREAMS の STALE 列で確認できます）です。未消費の Stream があると Snowflake が保持期間を一時的に延長してくれますが、上限は MAX_DATA_EXTENSION_TIME_IN_DAYS（既定 14 日）。失効した Stream は読めなくなり、再作成するしかありません。

ポイントは、パイプラインが動いている限り失効は起きないことです。差分があれば Task の DML が消費して、しおりが進みます。差分がなくても、WHEN での SYSTEM$STREAM_HAS_DATA の呼び出し自体が空の Stream を失効から守ります（公式ドキュメントに明記があります）。危ないのは ALTER TASK ... SUSPEND で Task を止めている間です。判定と消費のどちらも走らなくなり、そこから約 14 日で寿命が来ます。

踏みやすいのは、検証を中断・再開する環境です。パイプラインを止めて別の検証をして、2 週間後に戻ったら Stream が失効していた、が起こります。長期メンテナンスや障害対応で止める場合も同じ条件になります。

復旧は再作成だけで完結します。

-- オフセットは現時点にリセットされる
CREATE OR REPLACE STREAM FILE_CHANGES_STREAM ON TABLE RAW_FILES;

失効中に台帳へ入った変更は再作成後の Stream には乗りませんが、置き去りにはなりません。次に新しい変更が来て DAG が起動したとき、各 Task の差分処理は「未処理のもの全部」を対象にするため、失効中に溜まった分もまとめて拾われます（ここでもポイント 3 の設計が効きます）。待たずにすぐ流したい場合は、Task を介さず本体の処理を直接実行します。手動の EXECUTE TASK では取り込めないためですが、その理由はポイント 6 で後述します。

5. 稼働中の Task は作り直せない

Task の定義を変えたくなったとき、稼働中の Task には CREATE OR REPLACE できません。そこで構築スクリプト側を「消えた Task だけ作り直せる」形にしておくと、交換手順が単純になります。

-- ① ルートを止める（チェーン全体が起動しなくなる）
ALTER TASK PIPELINE_ROOT_TASK SUSPEND;

-- ② 変えたい Task だけ DROP
DROP TASK PARSE_TASK;

-- ③ 構築スクリプトを再実行
--    CREATE TASK IF NOT EXISTS で書いてあるので、消した Task だけが作り直される
--    スクリプト末尾の SYSTEM$TASK_DEPENDENTS_ENABLE で子→親の順に一括再開

構築スクリプトの Task 定義をすべて CREATE TASK IF NOT EXISTS にしておくのが肝です。CREATE OR REPLACE で書いてしまうと、スクリプト再実行のたびに全 Task が作り直され、SUSPEND 状態や実行履歴との整合を考えることが増えます。

なお、Task は作成直後、必ず SUSPENDED（停止）状態で生まれます。作り直した Task も、RESUME するまで動きません。再開時に ALTER TASK ... RESUME を 1 つずつ打つ必要はありません。SYSTEM$TASK_DEPENDENTS_ENABLE('<ルート Task 名>') を実行すると、手で止めたルートも含めて依存ツリーごと一括で有効化できます。

6. EXECUTE TASK はルートにしか効かない

途中の Task だけ失敗して、リトライも使い切ったとします。失敗した Task だけを再実行したくなりますが、EXECUTE TASK は DAG の子 Task には使えず、ルート（または単独 Task）にしか効きません。

そして Stream はルート起動時に消費済みなので、次に台帳へ変更が来るまで DAG は自動では再実行されません。失敗分がそのまま置き去りになります。

回収の第一手は、失敗地点から再開する EXECUTE TASK ... RETRY LAST です。直近の実行が失敗（FAILED / CANCELED）していて、その後タスクグラフを変更していない、といった条件はありますが、失敗直後に回収するという場面では自然に満たせます。

EXECUTE TASK PIPELINE_ROOT_TASK RETRY LAST;

当初は「条件を気にせず、ルートを手動起動して DAG 全体を流し直せばいい」と考えていました。ところがこれは動きません。手動の EXECUTE TASK でも WHEN 句は評価されるためです。実際に Stream が空（SYSTEM$STREAM_HAS_DATA が FALSE）の状態で EXECUTE TASK を実行してみました。実行履歴（INFORMATION_SCHEMA.TASK_HISTORY）の STATE は SKIPPED となり、本体の SQL まで到達していません。手動実行時に WHEN がどう扱われるかは公式ドキュメントに明記が見当たらず、実機で確認した結果です。失敗回収の場面では Stream はルート起動時に消費済みで空なので、素のルート手動起動は回収手段になりません。

RETRY LAST の条件を満たせない場合の最終手段は、Task を介さず本体の処理を直接実行することです。私のパイプラインでは各 Task の本体はストアドプロシージャの呼び出しなので、同じプロシージャを手で CALL すれば、WHEN 判定を通らずに同じ処理を実行できます。Task はスケジューラの皮にすぎないので、本体を単体で実行できる作りにしておけば回収手段に困りません。

どの回収手段でも安心材料になるのは、やはり差分ベースの設計です。仮に処理を頭から全部流し直しても、成功済みの分は対象ゼロで即終了し、実際に処理が走るのは失敗して未処理のまま残った分だけ。「どこから流しても壊れない」状態を保っておくことが、回収手順をいちばん単純にします。

7. タイムアウトは定常時の処理時間で決めない

USER_TASK_TIMEOUT_MS は Task の実行時間上限です。既定は 1 時間ですが、明示的に設定しています。AI 関数の呼び出しが返ってこないなど処理が想定外に終わらなくなったとき、サーバーレス Task は動いている時間のぶん課金され続けるためで、上限を設定しておけばそこで止まります。

気をつけたいのは初回の大量投入です。定常運用では数ファイルの差分処理が数分で終わっても、初回や一括追加では同じ Task が数百ファイルを処理することになり、定常時基準のタイムアウトだと正常な処理を誤って殺します。

私は AI 関数を含む重い Task だけ長め（60 分）、それ以外は 30 分に設定し、初回投入はファイルを分割して流しました。タイムアウトで切れても、差分ベースなら壊れません（未処理分が次回に回るだけ）。ただ、切れる前提の運用は健全でないため、大量投入が読めているなら分割するほうが安全です。

まとめ

振り返ると、7 つのうち 3・4・6・7 は「各処理を差分ベース・冪等に書いておく」ことで対処がシンプルになっています。Task の設計は Task 単体で閉じておらず、載せる処理の冪等性とセットで決まる。これが今回いちばんの学びでした。

おわりに

運用の落とし穴は踏んでから学ぶことが多いですが、この 7 つ分は先回りできるようにまとめたつもりです。どれか 1 つでも「先に知れてよかった」があれば嬉しいです。

参考

• Introduction to tasks | Snowflake Documentation
• Introduction to streams | Snowflake Documentation
• CREATE TASK | Snowflake Documentation
• SYSTEM$STREAM_HAS_DATA | Snowflake Documentation
• EXECUTE TASK | Snowflake Documentation