Connect

snowflake_streaming_pipe

License: This component requires an enterprise license. You can either upgrade to an Enterprise Edition license, or generate a trial license key that's valid for 30 days.

If the service only accepts connections from allowlisted IP addresses, add the public IP addresses that your Redpanda Connect hosts connect from, such as the address of your NAT gateway, to its allowlist. For pipelines on Redpanda Cloud, see Allow Pipeline Traffic Through External Firewalls.

This component is in beta. Its behavior might change in a future release.

Ingest data into Snowflake using Snowpipe Streaming, writing through a Snowflake pipe.

Introduced in version 4.115.0.

Ingest data into Snowflake using Snowpipe Streaming, writing through a Snowflake pipe rather than directly into a table.

Unlike snowflake_streaming, which opens channels on a table, this output opens them on a pipe, so any transformation defined in the pipe’s COPY INTO (stream-static join enrichment, WHERE filtering) runs before rows land. If you don’t need those, snowflake_streaming is the simpler choice: it creates its table for you and needs no pipe.

This is an Enterprise feature and requires a valid Redpanda Enterprise Edition license that includes Connect (licensing docs).

The pipe (and its table) must already exist; this output never issues CREATE PIPE or CREATE TABLE. A missing pipe is reported when the output connects, before any message is written. Fields with no matching column are dropped by Snowflake unless the table has ENABLE_SCHEMA_EVOLUTION = TRUE and the pipe uses MATCH_BY_COLUMN_NAME, in which case new fields become columns.

Moving an existing snowflake_streaming pipeline across is not a rename: several fields are removed and commit_timeout changes meaning. See the migration guide.

Data types

Each message is sent to Snowflake as one JSON row and Snowflake maps its fields to columns server-side; this output performs no type conversion of its own. What each column type accepts is therefore Snowflake’s rule, documented with the high-performance architecture’s limitations. Verified against a live account:

Snowflake column type JSON value

NUMBER, FLOAT

number, or a string holding one

BOOLEAN

boolean

CHAR, VARCHAR

string

TIMESTAMP_NTZ(9), TIMESTAMP_LTZ, DATE, TIME

RFC 3339 string; nanosecond precision is kept

VARIANT, OBJECT, ARRAY, MAP

any JSON value, including nested objects and arrays

A JSON null into a VARIANT column lands as a VARIANT null, not SQL NULL. GEOGRAPHY and GEOMETRY are not supported by Snowpipe Streaming.

Authentication can be configured using a RSA Key Pair.

Iceberg tables

Snowflake-managed Iceberg tables are supported as a target: Snowflake writes the Iceberg files server-side, so nothing in this output is table-type-specific, and partitioned tables work through a pipe whose COPY INTO reads FROM TABLE(DATA_SOURCE(TYPE ⇒ 'STREAMING')). Externally managed Iceberg catalogs are not supported by Snowpipe Streaming. Caveats seen on live tables:

  • VARCHAR(n) is not allowed on Iceberg tables; use VARCHAR.

  • VARIANT columns need an Iceberg v3 table (ICEBERG_VERSION = 3).

  • With server-side schema evolution, a new field holding a timestamp-like string can put the channel into ERR_CHANNEL_SCHEMA_EVOLUTION_ICEBERG_TIME_TYPE_SCALE_NOT_SUPPORTED. This output reports that as a permanent error naming the fix; pre-create timestamp columns with the type you want.

  • Each committed batch writes one file per table partition it touched, so a low-volume pipeline into a partitioned table produces many small files. Larger batches (see Request size and throughput) and a channel_name per source partition help.

Messages with an empty or whitespace-only body (Kafka/Redpanda tombstones, typically) are not sent: a Snowflake row has no representation for them. They are acknowledged upstream, counted in the snowflake_empty_rows_skipped metric and logged at debug level. If offset_token is set it is still evaluated for them, so use a metadata-based expression such as ${! @kafka_offset }, not one derived from the body. To land tombstones as rows, map them with a processor first.

Choosing a channel_name

By default every message shares one constant channel, <database>.<schema>.<pipe>. Set channel_name to split that up, most commonly one channel per Kafka/Redpanda partition.

The requirement: a channel must be written by one ordered source with at most one batch in flight. offset_token dedup compares each token against the channel’s single last-committed token and cannot tell "already committed by me" from "an unrelated writer got ahead of me", so two token sequences sharing a channel lose rows silently.

For a partitioned Kafka/Redpanda input, scope the channel per partition:

channel_name: "<pipe>-p${! @kafka_partition }"

This is safe at any max_in_flight only if the input dispatches one batch per partition at a time. The redpanda input does (unless unordered.enabled is set); kafka_franz and kafka do not by default (checkpoint_limit is 1024), so set checkpoint_limit: 1 there or max_in_flight: 1 here. If the input consumes several topics, include ${! @kafka_topic } in the name too. Never derive the name from the host: after a rebalance the new owner would start a channel with no committed history and re-send everything.

For a single ordered stream (Postgres logical replication, a single-partition topic): the default channel with max_in_flight: 1 is the simple choice. It is not safe with offset_token for a partitioned source, even at max_in_flight: 1: each partition is its own token sequence, so whichever is numerically ahead makes the others' rows look like duplicates. The linter rejects the common form of this (@kafka_offset with no @kafka_partition in the channel name); for anything else, scope per partition or leave offset_token unset.

If the requirement is violated anyway, a batch whose tokens start below the lowest token this process has already handled on the channel fails with an error naming the channel and tokens, and snowflake_out_of_order_submissions is incremented; the error recurs on every retry until channel_name or max_in_flight is fixed. A redelivery from a source that keeps one batch in flight per channel never trips it: such a source never goes back below where it started, however its batches are cut. The detection is one-sided and process-local: it catches a second source whose tokens lie below this process’s, but not one whose tokens lie above or between them, nor other replicas, nor anything after a restart; in those cases the rows would be dropped as duplicates with no error. Recover them from your dead-letter queue or the source; redelivery alone cannot re-insert them.

There is a limit of 10,000 streams per table — if using more than 10k streams please reach out to Snowflake support.

Performance

This output benefits from sending multiple messages in flight in parallel for improved performance. You can tune the max number of in flight messages (or message batches) with the field max_in_flight.

This output benefits from sending messages as a batch for improved performance. Batches can be formed at both the input and output level. You can find out more in this doc.

Request size and throughput

Each batch is one request per channel. Request bodies are zstd-compressed by default (compression), and Snowflake’s 4 MB per-request limit applies to the body as sent, after compression. A batch whose body is still over the limit is sent as several requests, split by message, each with its own offset-token range, so a large batch never fails for size; only a single message over the limit on its own cannot be sent (see tolerate_row_errors). Typical JSON compresses 5–8×, so 16–20 MB of messages usually fits in one request, and anything under about 3.5 MB always does. Memory per in-flight batch is roughly twice the uncompressed messages (the batch and the encoded body) plus the compressed body. Watch snowflake_compressed_output_size_bytes against the message bytes to see your ratio. Without offset_token, a batch sent as several requests can be duplicated if a later request fails after an earlier one committed: the whole batch is redelivered and the committed requests are sent again, so for large at-least-once batches prefer sizes that fit one request.

With the redpanda input, size batches at the input, through max_yield_batch_bytes (default 32KB, far below what one request carries), rather than with an output batching policy: an output policy pools every partition’s messages into one request, which serialises the partitions.

A channel carries one batch at a time. The next batch for a channel waits until the previous one has committed, so per-channel throughput is bounded by request size over commit latency (typically a few seconds in-region). Spread load across channels with a per-partition channel_name. Pipelining appends within a channel is planned.

  • Common

  • Advanced

output:
  label: ""
  snowflake_streaming_pipe:
    account: "" # No default (required)
    user: "" # No default (required)
    role: "" # No default (required)
    database: "" # No default (required)
    schema: "" # No default (required)
    pipe: "" # No default (required)
    private_key: "" # No default (optional)
    private_key_file: "" # No default (optional)
    private_key_pass: "" # No default (optional)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
    max_in_flight: 4
output:
  label: ""
  snowflake_streaming_pipe:
    account: "" # No default (required)
    url: "" # No default (optional)
    account_host: "" # No default (optional)
    user: "" # No default (required)
    role: "" # No default (required)
    database: "" # No default (required)
    schema: "" # No default (required)
    pipe: "" # No default (required)
    private_key: "" # No default (optional)
    private_key_file: "" # No default (optional)
    private_key_pass: "" # No default (optional)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)
    compression: zstd
    max_in_flight: 4
    channel_name: "" # No default (optional)
    offset_token: "" # No default (optional)
    commit_timeout: 60s
    commit_poll_interval: 250ms
    request_timeout: 1m0s
    channel_open_retry:
      max_attempts: 5
      initial_interval: 2s
      max_interval: 8s
    tolerate_row_errors: false

Fields

account

The Snowflake Account name. Which should be formatted as <orgname>-<account_name> where <orgname> is the name of your Snowflake organization and <account_name> is the unique name of your account within your organization.

Type: string

# Examples:
account: ORG-ACCOUNT

account_host

Override the control-plane host (default <account>.snowflakecomputing.com) while keeping ingest-hostname discovery. Use this for PrivateLink accounts. Ignored when url is set.

Type: string

# Examples:
account_host: my-account.privatelink.snowflakecomputing.com

batching

Configure a batching policy.

Type: object

# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m

batching.byte_size

The maximum total size (in bytes) that a batch can reach before it is flushed. When the combined size of all messages in the batch reaches or exceeds this limit, the batch is immediately sent to the next stage (such as a processor or output).

Set to 0 to disable size-based batching. When disabled, messages are flushed based on other conditions (such as count or period).

Type: int

Default: 0

batching.check

A Bloblang query that returns a boolean value indicating whether a message should end a batch.

Type: string

Default: ""

# Examples:
check: this.type == "end_of_transaction"

batching.count

The number of messages at which the batch is flushed. Set to 0 to disable count-based batching.

Type: int

Default: 0

batching.period

The length of time after which an incomplete batch is flushed regardless of its size. This field accepts Go duration format strings such as 100ms, 1s, or 5s. Supported time units are ns, us, ms, s, m, and h.

Type: string

Default: ""

# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms

batching.processors[]

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. All resulting messages are flushed as a single batch, so splitting the batch into smaller batches with these processors has no effect.

Type: array<processor>

# Examples:
processors:
  - archive:
      format: concatenate


# ---

processors:
  - archive:
      format: lines


# ---

processors:
  - archive:
      format: json_array

channel_name

The channel to write through. Resolved per message: a batch is grouped by channel name (preserving order within each group) and each group is appended, deduplicated when offset_token is set, and committed separately, so batch at the input so that a batch spans as few channels as possible. Two instances writing the same channel name conflict.

Defaults to a single constant channel, <database>.<schema>.<pipe>. See Choosing a channel_name for when that is safe and the per-partition pattern for Kafka/Redpanda inputs.

This field supports interpolation functions.

Type: string

# Examples:
channel_name: MYDB.PUBLIC.BETS_PIPE-p${!@kafka_partition}

channel_open_retry

Retry schedule for opening a channel and for the pipe pre-flight at connect time. The defaults allow about 22s of backoff. A channel open blocks this output’s other channels, so keep the budget modest.

Type: object

channel_open_retry.initial_interval

Wait before the first retry; each further retry doubles it.

Type: string

Default: 2s

channel_open_retry.max_attempts

Total attempts (including the first) before failing the batch. Only HTTP 429 and the 409 open-in-progress collision are retried.

Type: int

Default: 5

channel_open_retry.max_interval

Cap on the wait between retries.

Type: string

Default: 8s

commit_poll_interval

How often to poll channel status while waiting for a commit (see commit_timeout). Under backpressure from the status endpoint the interval doubles from this value up to 4s.

Type: string

Default: 250ms

commit_timeout

The maximum total time to wait for data to be committed to Snowflake before returning an error. If zero then no limit is used.

Type: string

Default: 60s

compression

How to compress each request body. Snowflake’s 4 MB per-request limit applies to the payload as sent, after compression, so zstd (Snowflake’s recommendation) lets a request carry several times more rows than none. A batch whose body is still over the limit is sent as several requests, split by message; a single message whose body is over the limit on its own is treated like a row Snowflake rejected (see tolerate_row_errors). The bytes sent are counted in snowflake_compressed_output_size_bytes.

Type: string

Default: zstd

Options: zstd, none

database

The Snowflake database to ingest data into.

Type: string

# Examples:
database: MY_DATABASE

max_in_flight

The maximum number of messages to have in flight at a given time. For outputs that send messages in batches, this limit applies to message batches. Increase this value to improve throughput.

Type: int

Default: 4

offset_token

The offset token used for exactly-once delivery. Each message’s token is compared to the channel’s last committed token (numerically when both are integers, otherwise byte-wise as strings); a message whose token is not newer is dropped as a duplicate. Delivery to the output must therefore be ordered, and messages within a batch must be in increasing token order; a retried batch may be dropped as duplicate if later batches succeeded meanwhile, so use a dead-letter queue for failures.

Non-integer tokens must be fixed-width for string order to be meaningful (0/F sorts after 0/10). postgres_cdc’s `@lsn already is; zero-pad anything you assemble yourself.

If unset, nothing is deduplicated and delivery is at-least-once; a retried batch spanning several channels re-appends on every channel that already landed.

Don’t add, remove, or change this expression on a pipeline that has already written to its channels. A channel’s tokens must stay one comparable sequence. A channel written without this field carries synthetic tokens (~rpcn-at-least-once-<digits>, visible as last_committed_offset_token) that compare newer than any real token, and a changed expression leaves new tokens judged against the old shape; this output refuses such batches where it can detect the mismatch (synthetic vs real, integer vs non-integer) and keeps refusing until you give the pipeline a fresh channel_name. Two different non-integer shapes cannot be told apart.

A duplicate (never a loss) is possible if an append is accepted but its commit is not confirmed within commit_timeout and the redelivered copy is appended before the original commits; raise commit_timeout (0 = no limit) to close that window.

For more information about offset tokens, see ^Snowflake Documentation

This field supports interpolation functions.

Type: string

# Examples:
offset_token: ${! @kafka_offset }

# ---

offset_token: postgres-${!@lsn}

pipe

The Snowflake pipe to ingest data into. It must already exist; a missing pipe is reported when the output connects. Unlike database and schema this value is not upper-cased, so match the stored identifier exactly (MY_PIPE for a pipe created as CREATE PIPE my_pipe). Snowflake’s managed default pipes (<TABLE>-STREAMING) only exist once a channel has been opened on the table through the table endpoint, which this output does not use, so a default pipe fails the connect-time check with ERR_PIPE_DOES_NOT_EXIST_OR_NOT_AUTHORIZED until something else has created it; create a named pipe instead.

Type: string

# Examples:
pipe: MY_PIPE

private_key

The PEM encoded private RSA key to use for authenticating with Snowflake. Either this or private_key_file must be specified.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

private_key_file

The file to load the private RSA key from. This should be a .p8 PEM encoded file. Either this or private_key must be specified.

Type: string

private_key_pass

The RSA key passphrase if the RSA key is encrypted.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

request_timeout

The timeout for a single HTTP request to Snowflake. A batch’s total wait is governed by commit_timeout.

Type: string

Default: 1m0s

role

The role to ingest as. Every streaming call is authorized against a session token scoped to this role (not the user’s default role), so it needs the required privileges on the pipe and table.

Type: string

# Examples:
role: SNOWPIPE_STREAMING_ROLE

schema

The Snowflake schema to ingest data into.

Type: string

# Examples:
schema: PUBLIC

tolerate_row_errors

Whether to tolerate rows Snowflake rejects after a batch is appended (malformed or uncastable values, reported asynchronously in the channel status). false (the default, matching the Kafka Connector’s errors.tolerance=none) fails the batch with an error naming the channel and Snowflake’s last error; true only logs a warning. Either way snowflake_rows_error_count is incremented and this output does not retry the rejected rows: Snowflake has already refused them and the redelivery dedupes to nothing, so failing the batch exists to alert an operator. To keep the rejected rows themselves, turn on error logging on the target table and Snowflake captures them in its error table; SNOWPIPE_STREAMING_CHANNEL_HISTORY has the error history with offsets (see error handling). When offset_token is unset such a rejection never fails the batch even with false, since without dedup a retry would only re-insert the good rows as duplicates.

The same policy applies to a message this output cannot send at all: one that spans several lines but is not valid JSON (multi-line bodies are compacted to one NDJSON row each, so they must parse; a body holding several newline-separated documents becomes that many rows), or one whose body alone is over the 4 MB request limit after compression. Such a message is dropped rather than sent, and treated like a row Snowflake rejected: the batch’s good rows land, snowflake_rows_error_count is incremented, and with offset_token set and false the batch fails once with the message number and parse error, after which its redelivery dedupes to nothing.

Type: bool

Default: false

url

Override the URL for both the control-plane and ingest-plane connections (default https://ORG-ACCOUNT.snowflakecomputing.com). This skips ingest-hostname discovery, so it is for local and test deployments only. For PrivateLink or other non-standard hostnames use account_host instead. If both are set, this field wins.

Type: string

# Examples:
url: http://localhost:8080

user

The user to run the Snowpipe Stream as. See Snowflake Documentation on how to create a user.

Type: string

Examples

Simple at-least-once ingestion

The minimal config: no offset_token or channel_name, so rows land at-least-once through a single channel named after the pipe, whatever the input. To move to exactly-once later, add offset_token together with a new channel_name (see `offset_token’s docs), scoped per partition if the input is partitioned.

input:
  generate:
    mapping: 'root = { "id": uuid_v4(), "amount": random_int(min: 1, max: 1000) }'
    interval: 1s
output:
  snowflake_streaming_pipe:
    account: "MYSNOW-ACCOUNT"
    user: MYUSER
    role: SNOWPIPE_STREAMING_ROLE
    database: "MYDATABASE"
    schema: "PUBLIC"
    pipe: "MY_PIPE"
    private_key_file: "my/private/key.p8"

Exactly once CDC into Snowflake

How to send data from a PostgreSQL table into Snowflake exactly once using Postgres Logical Replication.

A single ordered stream feeding the default channel, so max_in_flight: 1 is required (see channel_name’s docs). To run at-least-once instead, remove `offset_token (on a fresh channel_name if the pipeline has already run). 
input:
  postgres_cdc:
    dsn: postgres://foouser:foopass@localhost:5432/foodb
    schema: "public"
    slot_name: "my_repl_slot"
    tables: ["my_pg_table"]
    # Large batches perform best. A batch over the 4 MB compressed request
    # limit is sent as several requests, so count only needs to fit memory.
    batching:
      count: 50000
      period: 45s
    # One batch in flight at a time, so a retried batch never trails a newer
    # one into the channel.
    checkpoint_limit: 1
output:
  snowflake_streaming_pipe:
    # Postgres LSNs are fixed-width, so they order correctly as tokens.
    # Snapshot rows (stream_snapshot: true) carry no LSN; snapshot through a
    # separate at-least-once pipeline first.
    offset_token: "${!@lsn}"
    # Required with the default single channel; see channel_name's docs.
    max_in_flight: 1
    account: "MYSNOW-ACCOUNT"
    user: MYUSER
    role: SNOWPIPE_STREAMING_ROLE
    database: "MYDATABASE"
    schema: "PUBLIC"
    pipe: "MY_PG_PIPE"
    private_key_file: "my/private/key.p8"

Ingesting data exactly once from Redpanda

How to ingest data from Redpanda with consumer groups, decode the schema using the schema registry, then write the corresponding data into Snowflake exactly once through a pipe.

This is the pattern for any partitioned Kafka/Redpanda input: channel_name is scoped per partition, which is what makes exactly-once hold (see that field’s docs). To run at-least-once instead, remove offset_token (on a fresh channel_name if the pipeline has already run). 
input:
  redpanda:
    topics: ["my_topic_going_to_snow"]
    consumer_group: "redpanda_connect_to_snowflake"
    # We want very large batches - each batch will be sent to Snowflake individually
    # so to optimize query performance we want as big of files as we have memory for
    fetch_max_bytes: 100MiB
    fetch_min_bytes: 50MiB
    partition_buffer_bytes: 100MiB
pipeline:
  processors:
    - schema_registry_decode:
        url: "redpanda.example.com:8081"
        basic_auth:
          enabled: true
          username: MY_USER_NAME
          password: "${REDPANDA_SCHEMA_REGISTRY_PASSWORD}"
output:
  fallback:
    - snowflake_streaming_pipe:
        # One channel per partition, so partitions never share a channel;
        # the redpanda input keeps one batch per partition in flight, which
        # makes this safe at any max_in_flight.
        channel_name: "MYDB.PUBLIC.BETS_PIPE-p${!@kafka_partition}"
        # No batching policy here: batch size is governed by the input's
        # max_yield_batch_bytes. An output policy would pool partitions into
        # one request and serialise them.
        offset_token: ${! @kafka_offset }
        account: "MYSNOW-ACCOUNT"
        user: MYUSER
        role: SNOWPIPE_STREAMING_ROLE
        database: "MYDATABASE"
        schema: "PUBLIC"
        pipe: "MYPIPE"
        private_key_file: "my/private/key.p8"
    # In order to prevent delivery orders from messing with the order of delivered records
    # it's important that failures are immediately sent to a dead letter queue and not retried
    # to Snowflake. See the ordering documentation for the "redpanda" input for more details.
    - retry:
        output:
          redpanda:
            topic: "dead_letter_queue"

HTTP Server to push data to Snowflake

This example demonstrates how to create an HTTP server input that can receive HTTP PUT requests with JSON payloads, that are buffered locally then written to Snowflake in batches.

This example uses a buffer to respond to the HTTP request immediately, so it’s possible that failures to deliver data could result in data loss. See the documentation about buffers for more information, or remove the buffer entirely to respond to the HTTP request only once the data is written to Snowflake. 
input:
  http_server:
    path: /snowflake
buffer:
  memory:
    # Max inflight data before applying backpressure
    limit: 524288000 # 50MiB
    # Each flush is one batch. A batch over the 4 MB compressed request
    # limit is split into several requests, so this is a latency and
    # memory choice, not a hard cap.
    batch_policy:
      enabled: true
      byte_size: 2097152 # 2MiB
      period: "10s"
output:
  snowflake_streaming_pipe:
    account: "MYSNOW-ACCOUNT"
    user: MYUSER
    role: SNOWPIPE_STREAMING_ROLE
    database: "MYDATABASE"
    schema: "PUBLIC"
    pipe: "MYPIPE"
    private_key_file: "my/private/key.p8"
    # By default there is only a single channel per output pipe allowed
    # if we want to have multiple Redpanda Connect streams writing data
    # then we need a unique channel name per stream. We'll use the host
    # name to get a unique name in this example.
    channel_name: "snowflake-channel-for-${HOST}"