Snowpipe Streaming Guide: Low-Latency Ingestion and How It Differs from Snowpipe

Snowpipe Streaming inserts event data from applications and IoT devices directly into Snowflake tables at sub-second latency, without ever writing intermediate files. Where the original Snowpipe is a pull-based system triggered by file arrival, Snowpipe Streaming is a push-based streaming pipeline where the client actively sends row-level data.

Overview of Snowpipe Streaming

The streaming API in the Snowflake Ingest SDK is the heart of Snowpipe Streaming. A client application uses the SDK to open a logical write path called a channel, then sends row data as a Map through the insertRow API. The data is converted into micro-partitions on Snowflake's serverless infrastructure and applied to the target table. No stages to create, no COPY INTO to run.

Snowpipe vs Snowpipe Streaming Comparison

Both deliver automatic ingestion, but they differ fundamentally in input source, latency, and billing model.

Architecture

The Snowpipe Streaming data flow is a three-layer pipeline: client → channel → table. You can open multiple channels in parallel against a single table, and each channel tracks progress independently via its own offsetToken.

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

Implementation Example with the Java SDK

Here is a basic ingestion flow using the Snowflake Ingest SDK: create a client, open a channel, and push rows with 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);

Channel Management and offsetToken

Open a channel with openChannel and close it when you're done. Each channel represents an independent write stream to a table.

The offsetToken is a free-form string the client controls. For Kafka use the partition offset; for custom apps use a sequence number or UUID. On recovery, call getLatestCommittedOffsetToken to find the last commit point and resume from there — that's how you get ingestion with no duplicates and no gaps.

Implementing Exactly-Once Delivery

Snowflake guarantees at-least-once delivery, but with correct offsetToken management on the client you can build exactly-once semantics on top of it.

// 障害復旧時のリカバリパターン
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 Integration

The Snowflake Kafka Connector ships with a Snowpipe Streaming mode. Configure it as a Kafka Connect Sink Connector and events from Kafka topics flow automatically into Snowflake tables through 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"
}

Cost Model

Snowpipe Streaming is billed as serverless compute. You don't start a warehouse, but the ingestion work itself consumes Serverless Credits.

-- コスト確認クエリ
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;

What the Exam Tests

SnowPro Core and Data Engineer exams frequently test when to choose Snowpipe vs Snowpipe Streaming.

Sample Question

Frequently Asked Questions