Streams & Tasks で CDC/ELT パイプラインを組む

データ基盤の増分処理は、突き詰めると「変更をどう捕まえ、どのタイミングで、どの順番で流すか」に尽きます。SnowflakeのStreamはテーブルの変更行(CDC)を差分テーブルとして見せ、Taskはそれをスケジュールや依存関係に沿って自動実行します。バッチをシェルスクリプトやジョブスケジューラで束ねる代わりに、SQLとオブジェクトだけでELTパイプラインを宣言的に組めるのがStreamとTaskの強みです。本稿ではAWS(S3/Snowpipe)からの取り込みを起点に、StreamとTaskの正しい使い分けと、EMWが検証段階で先に潰しておいた落とし穴を、現場の実装指針としてまとめます。

01全体像：ロード→Stream→Taskで「宣言的な増分ELT」を組む

SnowflakeでのパイプラインはELT前提です。まずS3などから生データをそのままSnowflakeのRAWテーブルへロードし(Extract/Load)、変換(Transform)はSnowflakeの中でSQLとして実行します。この「変換」を全件洗い替えではなく、増えた分・変わった分だけ処理する増分パイプラインにする核が、StreamとTaskです。Streamがソーステーブルの変更行を差分として保持し、Taskがそれを定期的に、あるいは変更があった時だけ消費して下流のテーブルを更新します。ジョブネットのように処理を並べる代わりに、依存関係をオブジェクトとして宣言できるのが特徴です。

この形にすると、パイプラインの状態(どこまで処理したか)がStreamのオフセットとしてSnowflake側に持たれるため、外部にチェックポイントを持つ必要がありません。障害でリトライしても、消費済みの範囲を二重に処理しないのが構造的な安心材料です。

02Stream：変更を「差分テーブル」として捕まえる

Streamはソーステーブルに対するDML(INSERT・UPDATE・DELETE)を記録し、2つの時点の間で「何が変わったか」を行レベルの差分テーブルとして見せるオブジェクトです。これがSnowflakeにおけるCDC(Change Data Capture)の基本です。用途に応じて3タイプを使い分けます。標準(standard/delta)ストリームは挿入・更新・削除をすべて追跡し、挿入行と削除行を突き合わせて正味の差分を返します。append-onlyストリームは挿入行だけを返すためオーバーヘッドが小さく、追記中心のログ取り込みなどで高速です。insert-onlyストリームは外部テーブルおよび外部管理(externally managed)のIcebergテーブル向けで、ストレージ上のファイル削除は差分として扱いません。なお、Snowflake管理のIcebergテーブルには標準ストリームとappend-onlyストリームが利用できます。

Streamをクエリすると、ソース列に加えて3つのメタデータ列が付きます。METADATA$ACTION(INSERTかDELETEか)、METADATA$ISUPDATE(その変更がUPDATEの一部かどうか)、METADATA$ROW_ID(行を追跡する不変のID)です。UPDATEは内部的にDELETE+INSERTの組として表現されるため、METADATA$ISUPDATEを見ることで「更新」と「純粋な挿入・削除」を区別してMERGEを組み立てられます。

03オフセットと「消費」の作法：ここを外すと事故る

Streamで最も誤解されやすいのがオフセットの挙動です。Streamは内部にオフセット(読み取り位置)を持ち、DMLトランザクションでStreamを消費した時だけオフセットが現在位置まで前進します。つまりSELECT * FROM my_streamで中身を見るだけではオフセットは動かず、何度でも同じ差分が読めます。INSERT INTO ... SELECT ... FROM my_streamやMERGEのようにStreamをソースにしたDMLがコミットされて初めて、その差分は「消費済み」になり次回からは出てこなくなります。この消費はトランザクション整合的で、下流への書き込みが成功して初めてオフセットが進む点が信頼性の要です。

ここから2つの実装上の注意が導けます。1つ目は1つのStreamを複数の消費先で共有しないことです。ある消費先のDMLがオフセットを進めると、別の消費先はその差分を取りこぼします。複数の下流に同じ変更を配りたいなら、ソーステーブルに対して消費先ごとに別々のStreamを作るのが定石です。2つ目はStreamに新しいデータがあるかをSYSTEM$STREAM_HAS_DATAで確認できること。これを後述のTaskのWHEN句と組み合わせると、空のときは処理をスキップできます。

04Task：スケジュールと依存DAGで順番を保証する

TaskはSQL(単一ステートメント、ストアドプロシージャ、スクリプト)を定期実行する仕組みです。スケジュールはSCHEDULE='5 MINUTES'のような分指定か、SCHEDULE='USING CRON ...'のCRON式で指定します。単発のTaskだけでなく、ルートタスクと、その後にAFTER句でぶら下がる子タスクを連ねることで、有向非巡回グラフ(DAG)としての依存関係を宣言できます。同じ親を持つ子タスクは並列に、複数の親を持つ子タスクは全親の成功を待って動きます。仕上げ処理にはFINALIZEで定義するファイナライザタスクが使え、グラフ内の全タスクが成功・失敗いずれで終わっても最後に一度だけ走らせられます。1つのグラフあたり最大1,000タスク、1タスクあたり親・子それぞれ最大100という上限があり、グラフ内の全タスクは同一オーナー・同一データベース・同一スキーマに属する必要があります。

信頼性面では、連続失敗でタスクを自動停止させるSUSPEND_TASK_AFTER_NUM_FAILURESと、一時障害を吸収するTASK_AUTO_RETRY_ATTEMPTSを併用します。前者は「壊れたまま回り続けてクレジットを溶かす」事態を防ぐ安全弁として重要で、本番グラフには必ず設定しておきたいパラメータです。

05サーバーレスTaskとトリガードタスク：コストの勘所

Taskのコンピュートは2択です。ユーザー管理はWAREHOUSEを指定して自前のウェアハウスで実行します。サーバーレスはウェアハウスを指定せず、Snowflakeが必要なリソースを予測して自動割り当てします。サーバーレスは最大でXXLARGEウェアハウス相当までスケールし、SERVERLESS_TASK_MIN_STATEMENT_SIZE(既定XSMALL)とSERVERLESS_TASK_MAX_STATEMENT_SIZE(既定XXLARGE)でサイズの範囲を縛れます。課金はサーバーレスの計算リソース使用量(コンピュート時間)に基づき、秒単位で計測されます。少数のタスクを間欠的に回す用途や、スケジュール遵守を重視する場合はサーバーレスが向き、専用ウェアハウスを常時フル稼働させているような場合はユーザー管理が向きます。

さらに、変更ドリブンに寄せたいならトリガードタスクが有効です。SCHEDULEを持たず、WHEN SYSTEM$STREAM_HAS_DATA(...)で指定したStreamに変更が入った時だけ実行される形態で、サーバーレスとしてTARGET_COMPLETION_INTERVALを指定して使います。頻繁なポーリングが不要になり、変更が無い間はコンピュートを消費しないため、到着が読めないデータに対して低レイテンシと低コストを両立できます。

06検証で先に踏み抜いた3つの落とし穴

EMWは低コストの小規模導入から、大手飲料メーカー様・大手冷凍倉庫事業者様クラスの大規模まで複数業種でSnowflakeを設計・構築・運用してきました。そのなかでStreamとTaskは、検証(PoC)段階で必ず先に踏み抜いておくべき定番の落とし穴があります。いずれも本番投入前に潰し、重大障害0を継続しています。

(1) Streamの陳腐化。検証環境で、ソーステーブルのデータ保持期間内にStreamを消費しないと、Streamが陳腐化(stale)してCDCの差分を取りこぼすと分かりました。特に平日夜間しか変更が無いテーブルで、消費側のTaskを止めている間にオフセットが保持期間を追い越すと差分が失われます。本番では、STALE_AFTERを監視して閑散時間帯もTaskで空消費して回し、必要に応じてMAX_DATA_EXTENSION_TIME_IN_DAYSを見直すことで未然に防いでいます。

(2) 1Streamの多重消費。検証で、1つのStreamを複数のTaskから消費させると、先に走ったDMLがオフセットを進めてしまい、後続のTaskが空になる(差分をロストする)と再現しました。本番では消費先ごとにソーステーブル上へ別Streamを用意し、それぞれ独立したオフセットで安全に配れるようにしています。

(3) サーバーレス/短間隔スケジュールのコスト。検証で、専用ウェアハウスに短い間隔のスケジュールを組むと、変更が無い時間帯でもウェアハウスの起動・停止を繰り返し、そのたびに最小課金(60秒相当)が積み上がると分かりました。本番では、空振りが多い経路はトリガードタスク(サーバーレス)に切り替え、変更が無い間は課金されない構成にすることで、想定外のクレジット消費を未然に防いでいます。

07AWSとの接続点：S3ロードからパイプラインの入口まで

SnowflakeはAWS上で動くため、パイプラインの入口はS3との接続設計が肝になります。推奨はストレージ統合(storage integration)を作り、SnowflakeのIAMユーザーにS3バケットへのアクセスを許可するIAMロールを信頼させる方式です。これによりアクセスキーやシークレットといった静的な認証情報をSnowflake側に持たずに済みます。外部ステージをそのバケットに向け、S3のイベント通知(SQS経由)でSnowpipeが新規ファイルを検知して、サーバーレスにRAWテーブルへ継続ロードします。ここまでがELTの「L」で、その先をStreamとTaskで「T」として増分処理する、というのが全体の分担です。

セキュリティ面では、SnowflakeとAWS間の通信をパブリック経路に出さないAWS PrivateLink、S3側の保管データを保護するKMSによる暗号化を、要件に応じて組み合わせます。IAMロールによる最小権限、静的キーの排除、PrivateLinkによる経路の閉域化は、金融・製造・物流など監査要件の厳しい現場でそのまま効く基本設計です。

—まとめ：小さく確実に、増分で回す

StreamとTaskの本質は、「変更の捕捉(Stream)」と「実行順序と起動条件の宣言(Task)」を、外部のジョブ基盤に頼らずSnowflakeのオブジェクトとして持てることにあります。Streamのタイプとオフセットの作法を正しく押さえ、空振りはWHEN句とトリガードタスクでコンピュートを起こさず、失敗は自動停止とリトライで守る。そして陳腐化・多重消費・コストの3点は検証で先に潰しておく——この順番で組めば、増分ELTは驚くほど静かに、そして確実に回ります。まずは小さな一本のパイプラインから、宣言的な設計に置き換えていくのがおすすめです。