Snowpipe Streaming入門：低レイテンシ取り込みとSnowpipeとの違い

Snowpipe Streamingは、アプリケーションやIoTデバイスから発生するイベントデータを、 ファイルに書き出すことなくサブ秒レイテンシでSnowflakeテーブルへ直接挿入する仕組みです。 従来のSnowpipeがファイル到着をトリガとする「プル型」であるのに対し、 Snowpipe Streamingはクライアントが能動的に行データを送信する「プッシュ型」のストリーミング取り込みを実現します。

Snowpipe Streamingの概要

Snowpipe Streamingの中核はSnowflake Ingest SDKが提供するストリーミングAPIです。 クライアントアプリケーションはSDKを通じてチャネル（Channel）と呼ばれる論理的な書き込み経路を開き、 insertRow APIで行データをMapとして送信します。 送信されたデータはSnowflakeのサーバーレスインフラストラクチャ上でマイクロパーティションへ変換され、 ターゲットテーブルに反映されます。ステージの作成もCOPY INTOの実行も不要です。

Snowpipe vs Snowpipe Streaming 比較表

両者は「自動取り込み」という点では共通しますが、 入力ソース・レイテンシ・課金モデルに根本的な違いがあります。

アーキテクチャ

Snowpipe Streamingのデータフローは「クライアント → チャネル → テーブル」の3層構造です。 1つのテーブルに対して複数のチャネルを並列に開くことができ、 チャネルごとに独立したoffsetTokenで進捗管理されます。

┌─────────────────────┐
│   Client App        │  Snowflake Ingest SDK (Java)
│   / Kafka Connector │
└────────┬────────────┘
         │ insertRow(Map)
         ▼
┌─────────────────────┐
│   Channel           │  論理的な書き込み経路
│   (offsetToken管理) │  テーブルごとに複数並列可
└────────┬────────────┘
         │ HTTPS
         ▼
┌─────────────────────┐
│   Snowflake         │  サーバーレスインフラ
│   Micro-partition   │  行データ → マイクロパーティション変換
│   ┌───────────────┐ │
│   │ Target Table  │ │  数秒以内にクエリ可能
│   └───────────────┘ │
└─────────────────────┘

Java SDKによる実装例

Snowflake Ingest SDKを使った基本的な取り込みフローです。 クライアントを生成し、チャネルを開いてinsertRowで行を送信します。

// 1. ストリーミングクライアントの生成
Properties props = new Properties();
props.put("url", "https://<account>.snowflakecomputing.com");
props.put("user", "INGEST_USER");
props.put("private_key", privateKey);
props.put("role", "INGEST_ROLE");

SnowflakeStreamingIngestClient client =
    SnowflakeStreamingIngestClientFactory
        .builder("MY_CLIENT")
        .setProperties(props)
        .build();

// 2. チャネルのオープン
OpenChannelRequest req = OpenChannelRequest.builder("CH_ORDERS")
    .setDBName("PROD_DB")
    .setSchemaName("PUBLIC")
    .setTableName("ORDERS")
    .setOnErrorOption(OpenChannelRequest.OnErrorOption.CONTINUE)
    .build();

SnowflakeStreamingIngestChannel channel = client.openChannel(req);

// 3. 行の挿入
Map<String, Object> row = new HashMap<>();
row.put("ORDER_ID", 10001);
row.put("CUSTOMER", "Tanaka");
row.put("AMOUNT", 5800);
row.put("ORDER_TS", "2026-03-27T10:30:00Z");

InsertValidationResponse resp = channel.insertRow(row, "offset-10001");
if (resp.hasErrors()) {
    // エラーハンドリング
    resp.getInsertErrors().forEach(e ->
        System.err.println(e.getMessage()));
}

// 4. コミット済みオフセットの確認
String committed = channel.getLatestCommittedOffsetToken();
System.out.println("Last committed: " + committed);

チャネル管理とoffsetToken

チャネルはopenChannelで開き、不要になったらcloseします。 各チャネルはテーブルに対する独立した書き込みストリームを表します。

offsetTokenはクライアントが自由に設定できる文字列で、 Kafkaの場合はパーティションオフセット、独自アプリの場合は連番やUUIDを使います。 障害復旧時にgetLatestCommittedOffsetTokenで最後のコミット地点を取得し、 そこから再送することで重複や欠損のない取り込みを実現します。

Exactly-once保証の実装パターン

Snowflake側はAt-least-onceを保証しますが、 アプリケーション側でoffsetTokenを正しく管理することでExactly-once semanticsを構築できます。

// 障害復旧時のリカバリパターン
SnowflakeStreamingIngestChannel channel = client.openChannel(req);

// 前回のコミット済みオフセットを取得
String lastCommitted = channel.getLatestCommittedOffsetToken();

// Kafkaの場合: lastCommittedオフセット以降からコンシュームを再開
long resumeOffset = Long.parseLong(lastCommitted) + 1;
consumer.seek(partition, resumeOffset);

// 再開地点から取り込みを継続
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        Map<String, Object> row = parseRecord(record);
        channel.insertRow(row, String.valueOf(record.offset()));
    }
}

Kafka Connector連携

Snowflake Kafka ConnectorはSnowpipe Streamingモードをサポートしています。 Kafka ConnectのSink Connectorとして設定することで、 Kafkaトピックからのイベントを自動的にSnowpipe Streaming経由でテーブルへ取り込みます。

// Kafka Connector設定例（Snowpipe Streamingモード）
{
  "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
  "tasks.max": "4",
  "topics": "orders_topic",
  "snowflake.url.name": "<account>.snowflakecomputing.com",
  "snowflake.user.name": "KAFKA_USER",
  "snowflake.private.key": "${"
quot;}{file:/secrets/sf_key.pem}",
  "snowflake.role.name": "KAFKA_INGEST_ROLE",
  "snowflake.database.name": "PROD_DB",
  "snowflake.schema.name": "PUBLIC",
  "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
  "snowflake.enable.schematization": "true",
  "buffer.count.records": "10000",
  "buffer.flush.time": "10",
  "buffer.size.bytes": "5000000"
}

コストモデル

Snowpipe Streamingはサーバーレスコンピュートとして課金されます。 ユーザーがウェアハウスを起動する必要はありませんが、 取り込み処理に応じたServerless Creditが発生します。

-- コスト確認クエリ
SELECT
    START_TIME,
    END_TIME,
    CREDITS_USED,
    SERVICE_TYPE
FROM SNOWFLAKE.ACCOUNT_USAGE.METERING_HISTORY
WHERE SERVICE_TYPE = 'SNOWPIPE_STREAMING'
ORDER BY START_TIME DESC
LIMIT 20;

試験出題ポイント

SnowPro Core / Data Engineerでは、 SnowpipeとSnowpipe Streamingの使い分けが頻出テーマです。

サンプル問題

よくある質問