Streams & Tasks

StreamsとTasksは、Snowflakeでネイティブなデータパイプラインを構築するための中核機能です。Streamsがテーブルの変更差分（INSERT・UPDATE・DELETE）をキャプチャし、Tasksがその差分処理をスケジュール実行します。SnowPro試験ではData Engineering / Data Pipelineドメインで頻出のトピックです。

Streams（変更データキャプチャ）

Streamは、ソーステーブルに対するDML変更を追跡するオブジェクトです。内部的にオフセットを持ち、前回の消費位置から現在までの差分レコードを提供します。各行には3つのメタデータ列が付与されます。

-- Standard Stream（デフォルト：INSERT/UPDATE/DELETEすべて追跡）
CREATE OR REPLACE STREAM orders_stream
  ON TABLE raw.orders;

-- Append-only Stream（INSERTのみ追跡、UPDATEとDELETEは無視）
CREATE OR REPLACE STREAM orders_append_stream
  ON TABLE raw.orders
  APPEND_ONLY = TRUE;

-- Streamの差分を確認
SELECT * FROM orders_stream;

Stream種類の比較

CHANGES句によるTime Travel的な差分取得

CHANGES句を使うと、Streamオブジェクトを作成せずにテーブルの変更履歴をクエリできます。Streamと異なりオフセットを消費しないため、何度でも同じ差分を参照可能です。

-- 特定時刻以降の変更を取得
SELECT *
FROM raw.orders
  CHANGES (INFORMATION => DEFAULT)
  AT (TIMESTAMP => '2026-03-27 06:00:00'::TIMESTAMP_NTZ);

-- Statement IDを使った差分取得
SELECT *
FROM raw.orders
  CHANGES (INFORMATION => APPEND_ONLY)
  BEFORE (STATEMENT => '01b1a2b3-0601-1234-0000-abcdef012345');

Tasks（スケジュール実行）

Taskは、SQL文やストアドプロシージャを指定スケジュールで自動実行するオブジェクトです。CRON式による柔軟なスケジューリングと、分単位の間隔指定に対応しています。

-- CRON式によるTask（毎日AM6:00 JST）
CREATE OR REPLACE TASK daily_merge_task
  WAREHOUSE = ETL_WH
  SCHEDULE = 'USING CRON 0 6 * * * Asia/Tokyo'
  COMMENT = '日次の増分マージ処理'
AS
MERGE INTO mart.orders t
USING orders_stream s ON t.order_id = s.order_id
WHEN MATCHED AND METADATA$ACTION = 'DELETE' THEN DELETE
WHEN MATCHED AND METADATA$ISUPDATE = TRUE THEN
  UPDATE SET t.amount = s.amount, t.updated_at = CURRENT_TIMESTAMP()
WHEN NOT MATCHED AND METADATA$ACTION = 'INSERT' THEN
  INSERT (order_id, amount, created_at)
  VALUES (s.order_id, s.amount, CURRENT_TIMESTAMP());

-- Taskの有効化（作成直後はSUSPENDED状態）
ALTER TASK daily_merge_task RESUME;

SYSTEM$STREAM_HAS_DATAによる条件実行

WHEN SYSTEM$STREAM_HAS_DATA('stream_name')をTask定義に追加すると、Streamに未消費データがある場合のみTaskが実行されます。差分がないときの無駄なWarehouse起動を防止できます。

CREATE OR REPLACE TASK conditional_merge
  WAREHOUSE = ETL_WH
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  MERGE INTO mart.orders t USING orders_stream s
  ON t.order_id = s.order_id
  WHEN MATCHED THEN UPDATE SET t.amount = s.amount
  WHEN NOT MATCHED THEN INSERT VALUES (s.order_id, s.amount);

タスクツリー（DAG）

Tasksは親子関係を定義してDAG（有向非巡回グラフ）を構成できます。ルートTaskがスケジュールで起動し、子Taskは親の完了を待って順次実行されます。

-- ルートTask（スケジュールあり）
CREATE OR REPLACE TASK root_etl
  WAREHOUSE = ETL_WH
  SCHEDULE = '30 MINUTE'
AS SELECT 1;

-- 子Task（ルートの完了後に実行）
CREATE OR REPLACE TASK transform_orders
  WAREHOUSE = ETL_WH
  AFTER root_etl
AS
  INSERT INTO mart.orders_agg
  SELECT region, SUM(amount) FROM mart.orders GROUP BY region;

-- 孫Task（transform_ordersの完了後に実行）
CREATE OR REPLACE TASK notify_completion
  WAREHOUSE = ETL_WH
  AFTER transform_orders
AS
  CALL system$send_email(...);

-- DAG全体の有効化はルートTaskをRESUMEする
ALTER TASK root_etl RESUME;

Serverless Task

Enterprise Edition以上では、Warehouse指定の代わりにServerless Taskを利用できます。Snowflakeがコンピュートリソースを自動管理するため、Warehouseの起動・サイジング・コスト管理が不要になります。

CREATE OR REPLACE TASK serverless_etl
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  SCHEDULE = '10 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('events_stream')
AS
  INSERT INTO analytics.events_processed
  SELECT * FROM events_stream;

Dynamic Tablesとの比較

モニタリングと運用

• TASK_HISTORY：INFORMATION_SCHEMA.TASK_HISTORYテーブル関数でTaskの実行履歴・エラー詳細を確認
• STREAM_HAS_DATA：Streamに未消費データがあるかをBOOLEANで返す関数。Task内のWHEN句だけでなく、手動確認にも利用可能
• Stale Stream：Streamの保持期間（DATA_RETENTION_TIME_IN_DAYS + 14日）を超えると「stale」となり、消費不可になる。定期的な消費が必要
• 権限管理：Taskは所有ロールの権限で実行されるため、EXECUTE TASK権限の付与とロールの最小権限設計が重要

問題で確認

よくある質問