Snowflake Data HeroesPublicationへの投稿
❄️

Next-Gen Snowpipe Streaming：Snowflake次世代ストリーミング取り込みを理解する

yujmatsu
に公開
Snowflake
dataengineering
streaming
realtime
Snowpipe
tech

🚀 はじめに

（この記事は割と最近GAされた機能の紹介です）
Snowflakeで「ほぼリアルタイムにデータを入れたい」と思ったとき、選択肢はいくつかあります。

  • ファイル到着をトリガーに取り込む（Snowpipe / COPY INTO）
  • イベントやログをストリーミングで流し込む（Snowpipe Streaming）

そして最近、Snowflake公式ドキュメント上で 「Snowpipe Streaming: high-performance architecture」（本記事では便宜上「Next-Gen」と呼びます）という 新しい実装が登場し、ストリーミング取り込みの設計がグッと整理されました。

この記事では、以下のように整理します。

  • Snowpipe / Snowpipe Streaming の位置づけ
  • Next-Gen（high-performance） が何を変えたのか
  • GA（一般提供） のタイムライン（いつから本番利用しやすくなったか）
  • 設計・実装で押さえるべきポイント（PIPE / Channel / 監視 / コスト）
  • 「自分のユースケースだと何を選ぶべきか」

この記事の狙い
公式ドキュメントを読み始めたときに出てくる用語（PIPE / channel / schema validation / in-flight transformation…）を、最初の一周で腹落ちさせることです。

この記事でわかること

  • Snowpipe / Snowpipe Streaming / Next-Gen（high-performance）の違い
  • Next-Gen が GA になった流れ（AWS→Azure→GCP と段階展開）
  • Next-Genのキーワード：PIPE オブジェクト / サーバー側スキーマ検証 / COPY構文での in-flight 変換 / スループット課金
  • 最短で動かすための「ざっくり実装手順」（default pipe を使う）
  • 運用で困りがちな点（チャネル設計・スキーマ変更・監視）とベスプラ

まず整理：Snowpipe / Snowpipe Streaming は何が違う？

Snowpipe（ファイル到着ベース）

Snowpipeは「ステージ（S3/GCS/Azure Blob や internal stage）にファイルが置かれた」ことをトリガーに、Snowflakeへ自動ロードする仕組みです。

  • まとまったサイズのファイル（数MB〜数百MBなど）を対象にしやすい
  • 取り込みは “ファイル単位” で進む
  • 低遅延化はできるが、イベント1件ずつのような粒度だと設計が難しい

Snowpipe Streaming（イベント/行を継続取り込み）

Snowpipe Streamingは、アプリやストリーム基盤側から “行（レコード）” を連続的に Snowflake テーブルへ書き込むことを想定した仕組みです。

  • ログ、イベント、IoT、クリックストリームなど「常に流れてくるデータ」に強い
  • 取り込み → クエリ可能 までのレイテンシを小さくしやすい
  • クライアント側は SDK もしくは REST API で Snowflake に書き込む

図1: Snowflakeへの取り込み方式の全体像
図1: Snowpipeは「ファイル」、Snowpipe Streamingは「イベント/行」。ユースケースで選び分ける。

Next-Gen Snowpipe Streaming Architecture とは？

正式名称：Snowpipe Streaming: High-Performance Architecture

Snowflakeの公式ドキュメントでは、Next-Gen相当を 「Snowpipe Streaming: High-Performance Architecture」 として説明しています。

この“高性能アーキテクチャ”は、Snowpipe Streamingを 「大規模・リアルタイム取り込み（高スループット・低遅延）」 向けに作り直した実装です。

特徴を一言で言うと、

  • PIPE オブジェクトを中心に取り込み定義をサーバー側へ寄せる（=設定の集約）
  • スループット課金（取り込んだデータ量ベース） でコストが読みやすい
  • サーバー側スキーマ検証や COPY構文での in-flight 変換をサポート

という形で、「高頻度・高流量のストリーミング取り込み」を運用しやすくしています。

図2: Next-Gen（high-performance）アーキテクチャ概要
図2: クライアント → PIPE → バッファ → テーブル、という流れで「サーバー側に設定が集約」される。

いつ GA になった？

Next-Gen（High-Performance）アーキテクチャは、2025年にクラウド別に段階的にGA（一般提供）になりました。

  • 2025-09-23：AWSでGA
  • 2025-11-05：AzureでGA
  • 2025-11-10：GCPでGA

加えて、その後も「運用しやすくする改善」がリリースされました。

  • 2025-12-11：Default pipe（デフォルトパイプ）
  • 2025-12-17：Schema evolution（スキーマ進化）

ここでいうGA（General availability）は「プレビューではなく、本番利用を前提にした提供状態」を指します。
ただし機能の利用可否は リージョン/アカウント条件で変わることがあるため、必ず公式ドキュメント・リリースノートを確認してください。

図3: GAと主要アップデートのタイムライン
図3: 2025年後半にGA→運用機能（default pipe / schema evolution）が追随。

コストの考え方

取り込み方式の比較で、最初に混乱しがちなのが 「どこでクレジットがかかるの？」 です。
ポイントは 「取り込み（ingest）と、クエリ（query）は別」 ということです。

  • 取り込み（Snowpipe / Snowpipe Streaming）は サーバレス課金（=Warehouseを自分で立てない）
  • クエリ（SELECTして分析）は Warehouse課金（いつものSnowflake）

さらに Snowpipe Streaming はアーキテクチャで課金モデルが変わります。

  • Classic：サーバレス計算 + 接続時間などが課金の中心
  • Next-Gen（high-performance）：取り込んだデータ量（スループット） が課金の中心

Next-Genの具体例：

  • 取り込み課金はデータ量ベース（例：1TBあたり○クレジット）
  • 継続的な高流量を想定しているため、予測しやすいコストになる
  • Classic比で「接続維持コスト」が抑えられる傾向

詳細な料金体系はリージョン・契約により変動します。正確な見積もりには公式の「Pricing」ページと、Snowflakeアカウントチームへの確認をおすすめします。

用語ミニ辞典

ここだけ読めば、以降の説明がスムーズになります。

  • PIPE（パイプ）：取り込みの「設定（ルール）」を定義するオブジェクト。Next-Genではここが"中心"。
    ※ Snowpipe（ファイル）側にも PIPE があり、同名で混乱しやすいです（詳細は後述の「Q. Snowpipe（ファイル）のPIPEと、Streaming（High-Performance）のPIPEは同じ？」を参照）
  • Channel（チャネル）：クライアントがストリーミング書き込みを行う“接続・並列化の単位”。障害復旧や重複排除にも効く。
  • Default pipe：テーブルごとに自動で用意される「すぐ試せるPIPE」。まず動かす近道。
  • Schema validation：テーブルスキーマと合わないデータを（サーバー側で）弾く仕組み。
  • In-flight transformation：取り込みと同時に、軽い変換（例：型合わせ、カラムマッピング）をかけること。

Next-Genで重要になるキーワード：PIPE / Channel / Default pipe

1) PIPE オブジェクト（Next-Genの“中心”）

Next-Gen（high-performance）では、Snowpipe Streamingの取り込み設定を PIPEオブジェクトに集約します。

PIPEが担うイメージ：

  • ストリーミングデータの 入口（エントリポイント）
  • スキーマ検証や変換（COPY構文）などの サーバー側ルール
  • クライアントは「どのPIPE（またはどのテーブルのdefault pipe）に書くか」を指定してデータを送る

これにより、従来よりも クライアント側の責務（スキーマ整合の面倒）を減らしやすいのがポイントです。

図4: PIPEが“取り込みの定義レイヤー”になる
図4: 取り込みのルール（検証/変換）をPIPEに寄せると、クライアントはシンプルになる。

2) Channel（チャネル）

Snowpipe Streamingでは、クライアントが「チャネル」を開いてストリーミング書き込みを行います。

チャネルは実務上、次のような目的で重要です。

  • 並列化（スループットを出す）
  • 順序性・重複排除（オフセット管理）
  • 障害時の 再送/リカバリ の単位

ベスプラ（まずこれ）

  • Kafkaで言う「topic partition」などの概念と似た発想で、入力の分割単位＝チャネルを決めると運用が楽です。
  • 例：channel = topic + partition のように、安定した命名にする。

図5: チャネル設計のイメージ（分割・並列化）
図5: 「データの分割単位」をチャネルに対応させると、再処理や並列化が整理しやすい。

3) Default pipe（デフォルトパイプ）

Next-GenはPIPEが重要ですが、「じゃあ毎回 CREATE PIPE するの？」が最初の疑問になりがちです。

そこで登場するのが default pipe です。

  • テーブルごとに デフォルトのPIPE が提供される（Snowflakeが自動で生成）
  • 名前は <TARGET_TABLE_NAME>-STREAMING 形式（例：RAW_EVENTS-STREAMING）
  • 変換などの高度設定が不要なら、まずは default pipeで最短起動が可能

ただし、in-flight変換などをきっちり定義したい場合は、名前付きPIPEを自分で定義する方が良いケースがあります。

何がどう“次世代”なのか：Classic vs Next-Gen（high-performance）

Snowpipe Streamingには「classic architecture」と「high-performance architecture（Next-Gen）」があり、公式にはユースケースで使い分けが推奨されています。

ここでは初学者向けに、実務で効いてくる違いだけに絞って整理します。

図6: ClassicとNext-Genの違い（ざっくり比較）
図6: Next-Genは“PIPE中心・スループット課金・高流量向け”に寄せた設計。

選び方（迷ったら）

  • 常時高流量・リアルタイム分析が主目的
    → Next-Gen（high-performance）を第一候補

  • 流量が少ない / 断続的 / まずは軽く試したい
    → Classic（またはSnowpipe）も検討

「結局どれ？」の結論

  • ファイル中心なら Snowpipe
  • イベント中心なら Snowpipe Streaming
  • その中で “大規模・継続的” なら Next-Gen（high-performance）

Next-Genの機能概要（もう少し深掘り）

1) 高スループット＆低遅延

Next-Gen（high-performance）は、ストリーミング取り込みを 高スループット・低レイテンシにスケールさせることを主眼にしたアーキテクチャです。

初学者向けメモ
この領域は「データサイズ・カラム数・ネットワーク・チャネル数」などで実測が大きく変わります。
まずは チャンネル設計と監視を固めるのが再現性のある第一歩です。

2) SDK と REST API

Next-Genは、SDK（Java/Python など）と REST API を提供し、クライアント側からのストリーミング書き込みを可能にします（実装詳細は公式ドキュメントで確認してください）。

実務Tips

  • まずは SDK前提で設計する方がスループットを出しやすいことが多い
  • REST APIは「サーバレス実行環境から細いストリームを流す」などで便利（ただし要件次第）

3) in-flight transformations（COPY構文での変換）

Next-Genでは、PIPE側に COPY構文での変換定義を持たせられます。

ここが嬉しいのは、

  • クライアント側での変換ロジックを減らせる
  • 変換が“取り込みと一体”になり、運用が整理されやすい

一方で、取り込み時変換をやりすぎると「取り込み遅延」や「デバッグ難易度」が上がることもあります。

ベスプラ（おすすめ）

  • 取り込み時変換は「型合わせ」「カラムマッピング」など軽めに留める
  • 集計や複雑なJOINは downstream（Dynamic Tables / dbt / Tasks）へ寄せる

図7: 変換をどこでやるか（取り込み時 vs 下流）
図7: 取り込み時変換は軽めに。重い変換は下流レイヤーへ。

4) server-side schema validation（サーバー側スキーマ検証）

Next-Genはスキーマ検証をサーバー側へ寄せることで、クライアント側の複雑さを減らす方向です。

実務で効くポイント

  • スキーマの責務がサーバーに集約されると、「複数アプリが同じテーブルに書く」ケースでもガバナンスが効きやすい
  • 逆に、スキーマ変更の運用（いつどの順で変えるか）は重要になります

5) Schema evolution（スキーマ進化）

2025-12-17のアップデートで、Snowpipe Streaming（high-performance）における スキーマ進化のサポートが追加されました。

初学者向けの理解

  • ざっくり言うと「スキーマ差分を一定範囲で吸収できるようにする」方向の機能
  • ただし "なんでも自動でOK" ではないため、適用範囲・制約は公式を必ず確認してください

最低限これだけ（覚えるポイント）

  • 使うには、取り込み先テーブルで ENABLE_SCHEMA_EVOLUTION = TRUE を有効化する
  • サポート範囲に制約があります

具体例：

✅ サポートされる操作

  • 新しい列の追加（NULL許容）
  • データ型の拡張（例：INT → BIGINT、VARCHAR(10) → VARCHAR(20)）

❌ サポートされない操作

  • 列の削除
  • 列のデータ型変更（互換性のない変更）
  • 列幅の縮小（例：VARCHAR(20) → VARCHAR(10)）
  • 構造化型（OBJECT、ARRAY等）の変更は制限あり

実務Tips： スキーマ進化機能を過信せず、「計画的なスキーマ変更」と組み合わせて運用するのがベストプラクティスです。

最短で試す：Next-Genを触るための“ざっくり手順”

ここでは「イメージを掴む」ことを重視して、細かいコードよりも流れを整理します。

Step 0. 前提

  • Snowflakeアカウント（対象クラウド/リージョンでNext-Genが利用できること）
  • テーブル作成権限（DB/SCHEMA/TABLE）
  • Java または Python の実行環境（SDK利用）

Step 1. 取り込み先テーブルを作る

例（イベントログ想定）：

create or replace table raw_events (
  event_time timestamp_ntz,
  event_type string,
  user_id string,
  payload variant
);

Step 2. Default pipeを使う（最短ルート）

default pipe が利用可能な場合、テーブルに紐づく RAW_EVENTS-STREAMING のようなPIPEを使ってストリーミング書き込みできます。

  • 変換不要
  • クラスタリング不要
  • まずは動かす

という用途に向きます。

Step 3. クライアントでチャネルを開いて送る

Pythonの例（Snowflake Ingest SDKを使用）：

from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StreamingIngestClient
from datetime import datetime

# 接続設定（実際はconfigファイルや環境変数から読み込むのが推奨）
client = StreamingIngestClient(
    account="your_account",
    user="your_user",
    private_key_path="path/to/key.pem",
    role="your_role"
)

# チャネルを開く（チャネル名は安定した命名にする）
channel = client.open_channel(
    database="YOUR_DB",
    schema="YOUR_SCHEMA",
    table="RAW_EVENTS",
    channel_name="events_partition_00"  # 重要：安定した名前を使う
)

# データを送信
for event in event_stream:
    row = {
        "event_time": event["timestamp"],
        "event_type": event["type"],
        "user_id": event["user_id"],
        "payload": event["data"]
    }
    channel.insert_row(row)

    # 一定件数ごとにフラッシュ（パフォーマンスとレイテンシのバランス）
    if channel.get_buffered_row_count() >= 1000:
        channel.flush()

# 最終フラッシュとクローズ
channel.flush()
channel.close()

重要ポイント

  • チャネル名は安定していること：リトライや再起動のときに同じチャネルを使えるようにする
  • 適切なフラッシュ頻度：頻繁すぎると効率悪化、遅すぎるとレイテンシ増加
  • エラーハンドリング：本番では必ずリトライ・リカバリロジックを実装

Step 4. Snowflake側でクエリして確認

select event_type, count(*)
from raw_events
group by 1
order by 2 desc;

公式のチュートリアルも出ているのでこちらも参考にしてください。
https://docs.snowflake.com/ja/en/user-guide/snowpipe-streaming/snowpipe-streaming-high-performance-getting-started

Step 5. 監視（まずはチャネル状態を見る）

Next-Genではチャネル可視性が強化されているので、まずは

  • 送信できているか
  • 詰まっていないか
  • エラーが出ていないか

を確認する導線を作ります。

図8: 監視ループ（チャネル→アラート→復旧）
図8: 監視が無いと“止まっていても気づけない”。最初に監視ループを作る。

運用で効くベスプラ（Snowflake目線）

1) チャネル数は“多ければ良い”ではない

  • 少なすぎる → スループットが出ない
  • 多すぎる → クライアント管理が大変、追跡が難しい

まずは「上流の分割単位（partition/shard）」に合わせて設計し、ボトルネックが出たら調整するのがおすすめです。

2) スキーマ変更は"順番"が命

Next-Genではサーバー側スキーマ検証が中心になるため、スキーマ変更時は

  • 先にテーブル（/pipe）側を拡張
  • 次にクライアント側の送信を更新

のように、「受け口を広げてから送る」順番を守ると事故が減ります。

スキーマ進化機能を使う場合の注意点：

  • Schema evolution (ENABLE_SCHEMA_EVOLUTION = TRUE) を有効にしても、すべての変更が自動で吸収されるわけではない
  • サポートされる変更（列追加、型拡張等）とサポートされない変更（列削除、型縮小等）を理解しておく
  • 計画的なスキーマ変更と組み合わせて運用するのがベストプラクティス

3) in-flight変換は“軽く・少なく”

  • 取り込み時に大きな変換をやると、遅延・障害時の影響が増える
  • 変換は下流（dbt / Dynamic Tables / Tasks）へ寄せたほうが保守しやすいことが多い

4) 監視は「止まった検知」と「追いついている検知」を分ける

リアルタイム系は「遅れている」だけでもビジネス影響が出ます。

  • Hard fail：エラーで止まった
  • Soft fail：遅延が増えて追いついていない

この2系統を分けてアラート設計すると、運用が安定します。

よくある質問

Q. Snowpipe（ファイル）のPIPEと、Streaming（High-Performance）のPIPEは同じ？

結論：名前は同じ「PIPE」でも、用途が違う別物です。
初学者が混乱しやすいので、まずは次の2点だけ覚えると迷いにくくなります。

  • Snowpipe（ファイル）：ステージに置かれたファイルを COPY でロードするための「自動取り込み定義」
  • Snowpipe Streaming（高性能）：クライアント（SDK/REST）が行を送る入口となる「ストリーミング取り込み定義（検証/変換を含む）」

図9: 「PIPE」の混同ポイント（ファイル vs Streaming）
図9: SnowpipeのPIPEと、Streaming（High-Performance）のPIPEは"同名だが役割が違う"。

混同しがちなポイント（ここでつまずきがち）

  1. コマンドが似ている

    • どちらも SHOW PIPES / DESCRIBE PIPE で確認できるが、表示される情報が異なる
    • Snowpipe（ファイル）: AUTO_INGEST, NOTIFICATION_CHANNELなど
    • Snowpipe Streaming: AS_REPLICA, SCHEMA_EVOLUTION_ENABLEDなど

  2. 「COPY」の意味が違う

    • Snowpipe（ファイル）: ステージのファイルからテーブルへのロードを定義
    • Streaming（高性能）: 取り込み時の型合わせ/カラムマッピング等（in-flight transformation）

  3. 運用の単位が異なる

    • Snowpipe（ファイル）: ファイル単位で処理が進む
    • Streaming（高性能）: PIPE + Channelの組み合わせで運用（並列化/再送/重複排除の単位）

  4. トリガーが異なる

    • Snowpipe（ファイル）: ステージへのファイル到着イベント
    • Streaming（高性能）: クライアントからの連続的な行送信

見分け方の実践Tips

ドキュメントやコマンド実行時に「これはどちらのPIPE？」と迷ったら：

  • 「ステージ」「ファイル」「AUTO_INGEST」が出てくる → Snowpipe（ファイル）
  • 「Channel」「SDK」「Streaming」「スループット」が出てくる → Snowpipe Streaming

Q. Kafka Connector と何が違うの？

Kafka Connector（Snowflake Kafka Connector）も “Kafka→Snowflake” のパイプラインを作れますが、Snowpipe Streamingはより汎用のストリーミング取り込み基盤です。
「どのクライアント/基盤から入れるか」「運用主体はどこか」で選択肢が変わります。

  • Kafkaが中心で、Kafka運用基盤がある → Kafka Connector も候補
  • アプリから直接入れたい / Kafkaに依存したくない → Snowpipe Streaming が候補

Q. とりあえず試すなら何から？

まずは default pipe で「送る→入る→クエリできる」を確認するのが最短です。
そこで必要が出たら、名前付きPIPEで

  • 取り込み時変換（in-flight transformation）
  • 監査・ガバナンスを意識したスキーマ運用

に進むのがおすすめです。

😌 まとめ

  • Next-Gen Snowpipe Streamingは、公式には Snowpipe Streaming: high-performance architecture
  • PIPE中心の設計により、スキーマ検証・変換などのルールをサーバー側に集約しやすい
  • スループット課金により、継続的な高流量取り込みのコストが読みやすい
  • GAは2025年後半に段階展開（AWS→Azure→GCP）
  • まずは default pipe で最短起動し、必要に応じて名前付きPIPEへ

📚 参考リンク（必読）

Snowflake Data Heroes
Snowflake Data HeroesPublication

Snowflake データクラウドのユーザ会 SnowVillage のメンバーで運営しています。 Publication参加方法はこちらをご参照ください。 zenn.dev/dataheroes/articles/db5da0959b4bdd

Discussion