Creating an interactive table using streaming ingestion¶

Choosing an ingestion path¶

You can stream data into an interactive table by using any of the following paths, all of which sit on the Snowpipe Streaming high-performance architecture:

• Snowpipe Streaming SDK — for custom producers that use the Python or Java client directly. Best when you own the producer application and want the lowest-latency, lowest-overhead path into Snowflake.

• Snowflake Connector for Kafka — for teams that already run Apache Kafka or a Kafka-compatible service (Amazon MSK, Confluent Cloud). The connector calls the same Snowpipe Streaming v2 APIs under the hood.

• Snowflake Openflow — for managed, no-code pipelines. Use the PutSnowpipeStreaming2 processor to write records to Snowflake.

How streaming ingestion works: the default pipe¶

Every Snowpipe Streaming write goes through a PIPE object, which handles schema validation, in-flight transformations, and pre-clustering. For most interactive-table workloads, Snowflake creates and manages a default pipe for you — no CREATE PIPE needed.

CREATE INTERACTIVE TABLE does not create a pipe. Snowflake creates the default pipe on demand the first time a client opens a channel or calls get_pipe_info against the table. After that, it’s visible to SHOW PIPES and DESCRIBE PIPE.

The default pipe has the following characteristics:

• Name: <TABLE_NAME>-STREAMING. A table named ORDERS gets a default pipe named ORDERS-STREAMING.

• Snowflake-managed: You can’t run CREATE, ALTER, or DROP against the default pipe. Its OWNER attribute is NULL.

• Default ingestion semantics: It uses MATCH_BY_COLUMN_NAME to map incoming fields to columns. It doesn’t support in-flight transformations or pre-clustering.

Define your own named pipe when you need any of the following:

• In-flight transformations, such as projecting fields from $1:RECORD_CONTENT or calling SQL functions during ingest.

• Pre-clustering at ingest time (CLUSTER_AT_INGEST_TIME = TRUE).

• An explicit pipe you can grant on, monitor, and replicate independently from the table.

For the complete reference, see The PIPE object.

Required privileges for streaming ingestion¶

Before you run any of the ingestion options below, set up a role with the privileges needed to stream into the target table. Snowpipe Streaming also requires key-pair authentication; see Key-pair authentication and key-pair rotation for how to generate a key pair and associate it with the user.

The following example sets up a streaming_role for an interactive table named my_table whose default pipe is MY_TABLE-STREAMING:

CREATE ROLE streaming_role;
CREATE USER streaming_user;
GRANT ROLE streaming_role TO USER streaming_user;
GRANT USAGE ON DATABASE my_db TO ROLE streaming_role;
GRANT USAGE ON SCHEMA my_db.my_schema TO ROLE streaming_role;
GRANT INSERT, SELECT ON TABLE my_db.my_schema.my_table TO ROLE streaming_role;
GRANT OPERATE ON PIPE my_db.my_schema."MY_TABLE-STREAMING" TO ROLE streaming_role;

Option A: Using the Snowpipe Streaming SDK¶

The Snowpipe Streaming high-performance SDK lets your producer open a channel against the table’s pipes (default or custom) and append rows directly. This is the lowest-overhead path and is a good fit for custom producers.

Start by setting up a Python virtual environment and installing the SDK. Do not install packages system-wide with --break-system-packages.

python3 -m venv .venv
source .venv/bin/activate
pip install snowflake-ingest snowflake-connector-python

Create the interactive table in Snowflake:

CREATE OR REPLACE INTERACTIVE TABLE sensor_readings (
  device_id   INT,
  reading_ts  TIMESTAMP_NTZ,
  temperature NUMBER(6, 2)
)
CLUSTER BY (DATE_TRUNC('HOUR', reading_ts), device_id);

Create a profile.json file with your account, user, and private-key details. For the format and an example, see Step 2: Configure an authentication profile.

Then open a channel against the default pipe (SENSOR_READINGS-STREAMING) and append rows. The following example shows whta a minimal producer looks like:

import uuid
from datetime import datetime, timezone
from snowflake.ingest.streaming import StreamingIngestClient

client = StreamingIngestClient(
    client_name=f"sensor_demo_{uuid.uuid4().hex[:8]}",
    db_name="MY_DB",
    schema_name="MY_SCHEMA",
    pipe_name="SENSOR_READINGS-STREAMING",
    profile_json="profile.json",
)
channel, _ = client.open_channel("sensor_channel")

now = datetime.now(timezone.utc)
offset = 0
for device_id in range(1, 4):
    row = {
        "DEVICE_ID": device_id,
        "READING_TS": now.strftime("%Y-%m-%d %H:%M:%S.%f"),
        "TEMPERATURE": 68.5 + device_id,
    }
    offset += 1
    channel.append_row(row, str(offset))

channel.close()
client.close()

For the complete SDK reference, see Snowpipe Streaming key concepts and the getting-started tutorial.

Option B: Using the Snowflake Connector for Kafka¶

Use the Snowflake Connector for Kafka when you already run Kafka (or a Kafka-compatible service such as Amazon MSK or Confluent Cloud). The connector writes records through the Snowpipe Streaming v2 APIs, so the default pipe and custom-pipe rules described above apply.

CREATE OR REPLACE INTERACTIVE TABLE kafka_connector_demo (
  timestamp    TIMESTAMP_NTZ(6),
  country_code VARCHAR,
  url          VARCHAR
)
ENABLE_SCHEMA_EVOLUTION = TRUE
CLUSTER BY (TRUNC(timestamp, 'hour'));

{
  "name": "snowflake-interactive-sink",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
    "tasks.max": "1",
    "topics": "interactive_demo_topic",
    "snowflake.url.name": "<account>.snowflakecomputing.com:443",
    "snowflake.user.name": "<user>",
    "snowflake.private.key": "-----BEGIN PRIVATE KEY-----\n<your_private_key>\n-----END PRIVATE KEY-----",
    "snowflake.role.name": "STREAMING_ROLE",
    "snowflake.database.name": "MY_DB",
    "snowflake.schema.name": "MY_SCHEMA",
    "snowflake.warehouse.name": "<warehouse>",
    "snowflake.topic2table.map": "interactive_demo_topic:kafka_connector_demo",
    "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
    "snowflake.streaming.v2.enabled": "true",
    "snowflake.streaming.enable.altering.target.pipes.tables": "false",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "snowflake.enable.schematization": "true",
    "errors.tolerance": "all",
    "errors.log.enable": "true"
  }
}

"connector.class": "com.snowflake.kafka.connector.SnowflakeStreamingSinkConnector",
"snowflake.ingestion.method": "SNOWPIPE_STREAMING",
"snowflake.streaming.v2.enabled": "true",
"snowflake.streaming.enable.altering.target.pipes.tables": "false",
"snowflake.topic2table.map": "interactive_demo_topic:kafka_connector_demo",

python3 -m venv .venv
source .venv/bin/activate
pip install requests

# register_connector.py
import json
import sys
from pathlib import Path

import requests

connect_url = sys.argv[1].rstrip("/")
payload = json.loads(Path(sys.argv[2]).read_text(encoding="utf-8"))
name = payload["name"]

response = requests.put(
    f"{connect_url}/connectors/{name}/config",
    headers={"Content-Type": "application/json"},
    data=json.dumps(payload["config"]),
    timeout=30,
)
response.raise_for_status()
print(response.text)

python register_connector.py http://localhost:8083 snowflake-sink.json

Option C: Using Snowflake Openflow¶

Snowflake Openflow provides a managed, no-code pipeline for streaming data into Snowflake. Use the PutSnowpipeStreaming2 processor to write newline-delimited JSON records through the Snowpipe Streaming v2 API.

Openflow use the same default-pipe mechanism described in How streaming ingestion works: the default pipe to ingest data into Interactive tables. In the PutSnowpipeStreaming2 processor, point the Pipe Name property at either the table’s default pipe (<TABLE_NAME>-STREAMING) or a custom pipe that you created with a DATA_SOURCE(TYPE => 'STREAMING') COPY expression.

For the full list of Openflow processors, see All processors (alphabetical).

Monitoring data ingestion¶

The following Snowpipe Streaming view contains useful information to help you verify that data ingestion is happening to the interactive table and how much data is transferred:

• SNOWPIPE_STREAMING_FILE_MIGRATION_HISTORY view

Measuring end-to-end latency with row timestamps¶

Once the feature is enabled, you’ll be able to set ROW_TIMESTAMP = TRUE on the interactive table and use the row-level commit timestamp to measure end-to-end latency independently of any client-provided event time column. Latency then becomes a straightforward SQL expression:

-- Available once ROW_TIMESTAMP support for interactive tables is enabled.
ALTER TABLE kafka_connector_demo SET ROW_TIMESTAMP = TRUE;

SELECT AVG(DATEDIFF('millisecond', timestamp, ROW_TIMESTAMP(timestamp))) AS avg_latency_ms
  FROM kafka_connector_demo
  WHERE timestamp > DATEADD(MINUTE, -5, CURRENT_TIMESTAMP());

For general guidance on row timestamps, including how they behave across transactions and dynamic tables, see Use row timestamps to measure latency in your pipelines.

Managing data retention with storage lifecycle policies¶

Use a storage lifecycle policy to cap the amount of historical streaming data you keep. A common pattern for interactive tables is to retain a rolling window of the last two weeks.

The following policy expires rows whose ingest-time column is older than 14 days:

CREATE OR REPLACE STORAGE LIFECYCLE POLICY expire_policy_14d
  AS (c TIMESTAMP)
  RETURNS BOOLEAN ->
    c < DATEADD(DAY, -14, CURRENT_TIMESTAMP());

ALTER TABLE kafka_connector_demo
  ADD STORAGE LIFECYCLE POLICY expire_policy_14d ON (timestamp);

For the full set of options (archive tiers, tagging, monitoring), see Create and manage storage lifecycle policies.

Schema evolution for interactive tables¶

Interactive tables support schema evolution, which allows Snowflake to automatically add new columns when the incoming data contains fields that don’t yet exist in the table. This is particularly useful when streaming data from external systems like Kafka, where the source schema may change over time.

You enable schema evolution by setting ENABLE_SCHEMA_EVOLUTION = TRUE on the interactive table. You can set this property at table creation or on an existing table using ALTER TABLE.

In addition to automatic schema evolution, interactive tables support these manual ALTER TABLE operations:

• Adding new columns.

• Modifying NULL constraints on columns (SET NOT NULL or DROP NOT NULL).

CREATE OR REPLACE INTERACTIVE TABLE kafka_connector_demo (
  timestamp    TIMESTAMP_NTZ(6),
  country_code VARCHAR,
  url          VARCHAR
)
ENABLE_SCHEMA_EVOLUTION = TRUE
CLUSTER BY (TRUNC(timestamp, 'hour'));

ALTER TABLE kafka_connector_demo
  SET ENABLE_SCHEMA_EVOLUTION = TRUE;

ALTER TABLE kafka_connector_demo
  ADD COLUMN referrer VARCHAR;

ALTER TABLE kafka_connector_demo
  ALTER COLUMN country_code SET NOT NULL;

DESCRIBE TABLE kafka_connector_demo;