Snowflakeでデータ基盤を運用していると、「Snowflake内で増分処理を組みたい」場面が頻繁に出てきます。外部からのCDCではなく、Snowflake内のテーブル変更を追跡して下流テーブルに反映する、というニーズです。これを担うのがSnowflake StreamsとTasksの組み合わせです。
Streamsは「テーブルの変更を追跡するメタデータビュー」、Tasksは「スケジュール実行の仕組み」。両者を組み合わせると、Snowflake内で完結する増分パイプラインが組めます。実装パターンを整理します。
Streamsの基本動作
Streamsは「最後にチェックポイントしてから変わったレコード」を返すビューです。読み出すたびに、変更履歴がオフセットされます。
-- Streamの作成
CREATE STREAM stream_orders ON TABLE silver.orders;
-- 変更を取得
SELECT * FROM stream_orders;
-- 取得した結果に対して操作（INSERT/MERGE/UPDATE）を行うと
-- 自動的にチェックポイントが進むStreamの結果には`METADATA$ACTION`（INSERT/UPDATE/DELETE）、`METADATA$ISUPDATE`（更新かどうか）、`METADATA$ROW_ID`が追加されます。これらを使って下流の更新ロジックを組みます。
Standard vs Append-only Stream
| タイプ | 追跡対象 | 用途 |
|---|---|---|
| Standard Stream | INSERT・UPDATE・DELETE | 更新も削除も反映したい |
| Append-only Stream | INSERTのみ | イベントログのように追加だけ |
| Insert-only Stream（External Tables用） | INSERT検知 | 外部テーブル用 |
Append-onlyの方が軽量で性能が良いので、「追加だけ」と分かっているテーブルではこれを選びます。
Tasksとの組み合わせ
-- Streamの変更を定期的に処理するTask
CREATE TASK task_sync_orders
WAREHOUSE = WH_TRANSFORM
SCHEDULE = '5 MINUTE'
WHEN SYSTEM$STREAM_HAS_DATA('stream_orders')
AS
MERGE INTO gold.fct_orders t
USING (
SELECT * FROM stream_orders
WHERE METADATA$ACTION != 'DELETE'
) s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...;
-- Taskを起動
ALTER TASK task_sync_orders RESUME;`WHEN SYSTEM$STREAM_HAS_DATA(…)`で「変更があるときだけ実行」できるので、空処理を避けてコストを抑えられます。Snowflakeのウェアハウスの考え方はSnowflakeとはを参照してください。
dbt incrementalとの使い分け
| Streams + Tasks | dbt incremental | |
|---|---|---|
| 変更追跡 | Snowflake内部 | クエリで判定（updated_at等） |
| 遅延 | 1分〜 | スケジュール頻度に依存 |
| 削除の検知 | 標準でサポート | 「削除フラグ」設計が必要 |
| 運用ツール | SnowflakeのDDL | dbtプロジェクト |
| テスト・ドキュメント | 別途 | dbtで統合 |
「dbtプロジェクトで一元管理」ならincremental、「リアルタイム要件・削除検知」ならStreams、と分かれます。多くの組織でdbt incrementalから始め、ニーズが出てきたらStreamsを部分的に使う、という併用が現実的です。dbtの基本はincremental models設計を参照してください。
運用上のハマりどころ
- Stream保持期間：Streamの変更は標準で14日間保持。それより前にTaskが動かないとデータが失われる。重要な処理はTaskの正常稼働を監視。
- Stream Consumerは1つ：1つのStreamを複数のTaskが読むと、片方しかチェックポイントが進まない。複数下流が必要ならStreamを複数作る。
- 大きなテーブルへのStream作成：既存の大規模テーブルにStreamを張ると、初回コストが大きい。導入時のテーブル運用に注意。
- Taskのコスト管理：頻繁にTaskを動かすとウェアハウスのクレジットが嵩む。`STREAM_HAS_DATA`で無駄な起動を抑える。
典型的な利用パターン
- BronzeからSilverへの増分反映：Streamで変更検知、MERGEで反映
- SilverからGoldの集計表更新：Streamで変更検知、集計テーブルを増分更新
- 監査ログの自動構築：Streamで全変更を記録、別テーブルに追記
- 外部システムへの通知：Streamで変更検知、外部関数でSNS/API呼び出し
まとめ
- Streamsはテーブル変更を追跡するメタデータビュー、Tasksはスケジュール実行。
- 組み合わせでSnowflake内完結のCDCパイプラインが組める。
- Standard / Append-onlyの2タイプで、用途に応じて選ぶ。
- dbt incrementalとは「内部追跡 vs クエリ判定」で使い分け。
- 保持期間14日、1Stream=1Consumer、大規模テーブル初回コスト等が運用注意点。
全体像はCDC実装ガイド、隣接はDebezium とは・PostgreSQL論理レプリケーション。Snowflake内CDC設計の壁打ちは、DE-STKの初回相談（30分・無料）もご利用ください。
よくある質問（FAQ）
Q. Dynamic Tablesとどう違いますか？
A. Dynamic TablesはSnowflakeの新機能で、「SQLで定義した結果テーブルが自動的に最新に保たれる」マテリアライズドビュー的な機能です。StreamsとTasksを使う代わりに、Dynamic Tablesで宣言的にパイプラインを書く選択肢が増えています。Streams+Tasksは「自分で更新ロジックを書く」のに対し、Dynamic Tablesは「Snowflakeが自動で最適化」します。と棲み分けます。
Q. dbtからStreamsを使えますか？
A. dbtの`pre_hook`/`post_hook`でStreamのCRUDを操作できますが、dbtプロジェクトのモデル管理とStreamsの状態管理が分かれることに注意が必要です。dbtコミュニティ製のパッケージも存在しますが、多くの組織は「dbt incremental＋必要に応じてStreamsを別運用」のハイブリッドにしています。
Q. Streamの保持期間14日を伸ばせますか？
A. デフォルト14日は、ソーステーブルの`DATA_RETENTION_TIME_IN_DAYS`に依存します。これを伸ばすと、Streamの保持期間も伸びます。ただしストレージコストが増えるので、業務要件で必要な期間に絞ります。「Taskが3日止まっても大丈夫」なら7日設定、というように設計します。
Q. Snowflake外部にイベントを流したい場合は？
A. External Functions（AWS Lambda・Azure Functions経由）でAPI呼び出し、Snowflake Stream → External StageへのCOPYでファイル経由配信、などの方法があります。本格的なイベント駆動アーキテクチャを組むなら、Kafkaを別に立ててDebezium+Kafkaで構成する方が素直です。