Connect

Migrate from snowflake_streaming to snowflake_streaming_pipe

Move a pipeline from the snowflake_streaming output to the snowflake_streaming_pipe output when you want Snowflake to transform rows before they land, for example with a join or a WHERE filter in the pipe’s COPY INTO statement. The two outputs are configured differently, so a migration is more than a rename.

After reading this page, you will be able to:

  • Create the Snowflake table and pipe that snowflake_streaming_pipe writes through

  • Translate a snowflake_streaming configuration field by field

  • Cut a running pipeline over to snowflake_streaming_pipe

Prerequisites

Choose between the two outputs

Keep snowflake_streaming if you don’t need a pipe. It writes directly to a table, can create and evolve the table for you, and can run a Bloblang mapping on each message.

Use snowflake_streaming_pipe if you want the transformation to run in Snowflake. Each message is sent as one JSON row, Snowflake maps its fields to columns, and the pipe’s COPY INTO statement runs before the rows land.

The snowflake_streaming_pipe output is in beta, while snowflake_streaming is stable. Test the new output before you move a production pipeline to it.

Create the table and pipe

The snowflake_streaming_pipe output never creates a table or a pipe. It checks that the pipe exists when it connects and fails before writing any message if it doesn’t.

  1. Create the target table, or reuse the table your snowflake_streaming output writes to.

  2. To add a column when a message contains a new field, the job schema_evolution did in snowflake_streaming, enable schema evolution on the table:

    ALTER TABLE MY_DATABASE.PUBLIC.EVENTS SET ENABLE_SCHEMA_EVOLUTION = TRUE;

  3. Create a named pipe that copies streamed rows into the table. MATCH_BY_COLUMN_NAME maps each JSON field to the column of the same name, and is also required for schema evolution:

    CREATE PIPE MY_DATABASE.PUBLIC.EVENTS_PIPE AS
  COPY INTO MY_DATABASE.PUBLIC.EVENTS
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

    Don’t rely on Snowflake’s managed default pipe (<TABLE>-STREAMING). It only exists after something opens a channel on the table through the table endpoint, which snowflake_streaming_pipe doesn’t use.

Translate the configuration

These fields work the same way in both outputs and carry over unchanged: account, user, role, database, schema, private_key, private_key_file, private_key_pass, batching, max_in_flight, and offset_token.

The rest change as follows.

snowflake_streaming field In snowflake_streaming_pipe

table

Replaced by pipe, the name of the pipe you created. Unlike database and schema, pipe isn’t upper-cased, so match the stored identifier exactly. The pipe field doesn’t support interpolation, so use one output for each pipe.

url

Kept, but for local and test deployments only: it also overrides the ingest host, which snowflake_streaming_pipe otherwise discovers. If you set url for a custom or PrivateLink hostname, use account_host instead.

channel_prefix

Removed. Use channel_name. See Channel names.

channel_name

Kept, but resolved for every message instead of only the first message in a batch. See Channel names.

commit_backoff, commit_timeout

Replaced by a single commit_timeout deadline and a commit_poll_interval. See commit_timeout.

mapping

Removed. Run the mapping in a mapping processor before the output, or move the transformation into the pipe’s COPY INTO statement.

schema_evolution

Removed. Set ENABLE_SCHEMA_EVOLUTION = TRUE on the table and use MATCH_BY_COLUMN_NAME in the pipe, as shown in Create the table and pipe.

init_statement

Removed. Create the table and pipe before you start the pipeline.

message_format

Removed. Each message must be a JSON object whose keys are column names. If your messages are arrays (message_format: array), convert them to objects in a mapping processor.

timestamp_format

Removed. Snowflake parses timestamp columns from RFC 3339 strings, so send timestamps in that format.

build_options

Removed. The output sends JSON and Snowflake converts the values to column types, so there is no local build step to tune.

The snowflake_streaming_pipe output also adds fields with no snowflake_streaming equivalent. The defaults suit most pipelines:

  • compression: compresses each request (zstd by default), so more rows fit under Snowflake’s 4 MB request limit.

  • tolerate_row_errors: whether a row Snowflake rejects fails the batch (false, the default) or only logs a warning. A rejected row fails the batch only when offset_token is set.

  • request_timeout: the timeout for a single HTTP request.

  • channel_open_retry: the retry schedule for opening a channel.

  • account_host: the control-plane host to use, for example for PrivateLink accounts. Unlike url, it keeps ingest-hostname discovery.

Channel names

The snowflake_streaming output interpolates channel_name on the first message of each batch, so every message in a batch is written to the same channel. The snowflake_streaming_pipe output resolves channel_name for each message, groups the batch by channel, and commits each group separately. If you set channel_prefix, replace it with a channel_name that gives each writer its own channel.

Each channel must be written by one ordered source. For a partitioned Kafka or Redpanda input, include the partition in the name:

channel_name: "EVENTS_PIPE-p${! @kafka_partition }"
A per-partition channel is safe at any max_in_flight only if the input sends one batch per partition at a time. The redpanda input does this unless unordered_processing.enabled is true. The kafka_franz and kafka inputs don’t by default, because checkpoint_limit defaults to 1024. With offset_token set, set checkpoint_limit: 1 on those inputs or max_in_flight: 1 on the output. Otherwise, Snowflake can silently drop rows as duplicates.

The linter rejects an offset_token that uses @kafka_offset unless channel_name includes @kafka_partition. It also rejects offset_token with the default single channel and a max_in_flight above 1. For the full rules, see Choosing a channel_name.

commit_timeout

In snowflake_streaming, commit_timeout is deprecated. It sets the same limit as commit_backoff.max_elapsed_time and overrides it, and commit_backoff controls how often the output checks for a commit.

In snowflake_streaming_pipe, commit_timeout is the only deadline. It is the total time to wait for a batch to commit before the batch fails, and 0 means no limit. The output checks every commit_poll_interval (250ms by default) and backs off only when Snowflake pushes back.

To translate:

  1. Set commit_timeout to your old commit_timeout, or to commit_backoff.max_elapsed_time if commit_timeout wasn’t set. The old commit_backoff.max_elapsed_time and the new commit_timeout both default to 60s.

  2. Remove the other commit_backoff fields. No equivalent exists for initial_interval, max_interval, or multiplier.

With offset_token set, a batch whose commit isn’t confirmed within commit_timeout can be written twice if it’s redelivered before the original commits. Raise commit_timeout to make that less likely.

Examples

This snowflake_streaming output reads from a partitioned Redpanda topic and writes to the EVENTS table:

output:
  snowflake_streaming:
    account: MYORG-MYACCOUNT
    user: MYUSER
    role: MYROLE
    database: MY_DATABASE
    schema: PUBLIC
    table: EVENTS
    private_key_file: ./rsa_key.p8
    channel_name: "events-p${! @kafka_partition }"
    offset_token: "${! @kafka_offset }"
    schema_evolution:
      enabled: true
    commit_backoff:
      max_elapsed_time: 120s

The same pipeline with snowflake_streaming_pipe, writing through the EVENTS_PIPE pipe created earlier:

output:
  snowflake_streaming_pipe:
    account: MYORG-MYACCOUNT
    user: MYUSER
    role: MYROLE
    database: MY_DATABASE
    schema: PUBLIC
    pipe: EVENTS_PIPE
    private_key_file: ./rsa_key.p8
    channel_name: "EVENTS_PIPE-p${! @kafka_partition }"
    offset_token: "${! @kafka_offset }"
    commit_timeout: 120s

Schema evolution now comes from the table and pipe settings, so the schema_evolution block is gone.

Cut over

  1. Create the table and pipe, as described in Create the table and pipe.

  2. Translate the output configuration and lint it:

    rpk connect lint ./pipeline.yaml

  3. Stop the pipeline that runs snowflake_streaming.

  4. Start the pipeline with snowflake_streaming_pipe. Keep the same input configuration, including the consumer group, so the input resumes from the offsets the old pipeline committed.

  5. Check that rows arrive in the table, and watch the snowflake_rows_error_count metric for rows Snowflake rejects.

Next steps