目次 › 第3章 › 3-4

第3章 データロード・アンロードと接続 / 想定学習時間:30〜40分 / 最終確認:2026年8月

3-4. StreamsとTasks(CDC・タスクグラフ・サーバーレスタスク)

1. Stream:テーブルの「変更」を追跡するオブジェクト

「昨夜の取り込み以降に変わった行だけを下流のテーブルに反映したい」——このような差分処理を実現するのが Stream(ストリーム)です。Stream は、対象テーブルに対する変更(挿入・更新・削除)を追跡するオブジェクトで、いわゆる CDC(Change Data Capture)を Snowflake の中で完結させます。

-- テーブルの変更を追跡するストリームを作成する
CREATE STREAM sales_stream ON TABLE raw_sales;

-- ストリームを SELECT すると「前回のオフセット以降の変更行」が見える
SELECT * FROM sales_stream;

Stream を SELECT すると、変更された行に加えて、変更の種類を示すメタデータ列が付いてきます。

2. Stream の3つの種類

3. オフセットの前進と stale 化

Stream は「どこまでの変更を読んだか」を示すオフセットを内部に持っています。重要なのはオフセットが進むタイミングです。

• Stream を SELECT しただけではオフセットは進みません。何度読んでも同じ変更セットが見えます。
• Stream を DML(INSERT ... SELECT、MERGE など)のソースとして消費したとき、そのトランザクションの成功と同時にオフセットが前進し、消費済みの変更は Stream から消えます。

-- 1回目:変更行が見える
SELECT COUNT(*) FROM sales_stream;   -- 例:120行

-- SELECT だけならオフセットは進まないので、もう一度見ても同じ
SELECT COUNT(*) FROM sales_stream;   -- 120行のまま

-- DML で消費するとオフセットが前進する
INSERT INTO sales_clean
SELECT sale_id, amount
FROM sales_stream
WHERE METADATA$ACTION = 'INSERT';

-- 消費後は空になる(新しい変更が来るまで)
SELECT COUNT(*) FROM sales_stream;   -- 0行

もう1つの注意点が stale(失効)化です。Stream の差分情報は、元テーブルのデータ保持期間(Time Travel の保持期間)に依存しています。Stream を長期間消費しないまま、オフセットが保持期間の外側に取り残されると、その Stream は stale となり、差分を読めなくなります。定期的に消費するか、保持期間を適切に設定することが防止策です(保持期間は 2-2 参照)。

4. Task:SQL をスケジュール実行する

Task(タスク)は、SQL 文(1つ)をスケジュールに従って自動実行するオブジェクトです。スケジュールの指定方法は2つあります。

-- 方法1:分間隔で指定する
CREATE TASK load_task
  WAREHOUSE = etl_wh
  SCHEDULE = '5 MINUTE'
  AS
  INSERT INTO sales_clean SELECT ... ;

-- 方法2:CRON 式で指定する(毎日午前2時、東京時間)
CREATE TASK nightly_task
  WAREHOUSE = etl_wh
  SCHEDULE = 'USING CRON 0 2 * * * Asia/Tokyo'
  AS
  CALL nightly_proc();

実行に使うコンピュートは2方式から選べます。

また、タスクは作成した直後は suspended(停止)状態です。ALTER TASK ... RESUME を実行して初めてスケジュールどおりに動き始めます。デバッグなどで即時に1回だけ動かしたいときは EXECUTE TASK を使います。

5. タスクグラフ(DAG)と Stream との組み合わせ

複数のタスクは AFTER 句でつないでタスクグラフ(DAG)を構成できます。ルールは次のとおりです。

• スケジュールを持てるのはルートタスクだけ。後続タスクは AFTER 先行タスク名 で連結し、先行タスクの完了を受けて実行されます。
• グラフ全体を動かすには、後続タスクから先に RESUME し、最後にルートタスクを RESUME するのが基本手順です(まとめて有効化する SYSTEM$TASK_DEPENDENTS_ENABLE もあります)。

さらに、WHEN 句に SYSTEM$STREAM_HAS_DATA('stream名') を指定すると、「ストリームに未消費の変更があるときだけ」タスク本体を実行できます。変更がなければウェアハウスを起動せずにスキップするため、無駄なコンピュートコストを避けられます。

-- ルートタスク:ストリームにデータがあるときだけ実行する
CREATE TASK merge_task
  WAREHOUSE = etl_wh
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('sales_stream')
  AS
  MERGE INTO sales_clean t
  USING sales_stream s ON t.sale_id = s.sale_id
  WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' THEN DELETE
  WHEN MATCHED THEN UPDATE SET t.amount = s.amount
  WHEN NOT MATCHED THEN INSERT (sale_id, amount) VALUES (s.sale_id, s.amount);

-- 後続タスク:merge_task の完了後に集計を更新する
CREATE TASK agg_task
  WAREHOUSE = etl_wh
  AFTER merge_task
  AS
  INSERT OVERWRITE INTO sales_daily SELECT ... ;

-- 有効化:後続 → ルートの順に RESUME する
ALTER TASK agg_task RESUME;
ALTER TASK merge_task RESUME;

練習問題