Skip to main content

snowflake_stream output

Streams JSON messages into a Snowflake table with Snowpipe Streaming.

# Common config fields, showing default values
output:
  label: ""
  snowflake_stream:
    account: "" # No default (required)
    user: "" # No default (required)
    role: "" # No default (optional)
    database: "" # No default (required)
    schema: "" # No default (required)
    table: "" # No default (required)
    pipe: "" # No default (optional)
    private_key: "" # No default (optional)
    private_key_file: "" # No default (optional)
    private_key_pass: "" # No default (optional)
    mapping: "" # No default (optional)
    channel_prefix: "" # No default (optional)
    offset_token: "" # No default (optional)
    max_in_flight: 0 # No default (optional)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""

Sends each batch of JSON objects to a Snowflake table over the Snowpipe Streaming REST API, with low latency and without a virtual warehouse. Each message must be a JSON object whose keys match the table's column names; Snowflake converts JSON values to the column types on ingest.

Authentication uses key-pair (JWT) auth: register the public key on the user with ALTER USER ... SET RSA_PUBLIC_KEY and give this output the private key.

Rows are streamed through a pipe: by default the table's own streaming pipe, so only the table has to exist. To let new JSON fields add columns automatically, enable schema evolution on the table itself (ALTER TABLE ... SET ENABLE_SCHEMA_EVOLUTION = TRUE).

A batch is acknowledged once Snowflake reports it committed, so a failure before that point retries the batch. Without offset_token delivery is at-least-once; with it, rows the channel has already committed are skipped, giving exactly-once delivery across restarts. Rows Snowflake accepts but cannot load (for example a value that does not convert to its column type) are logged and counted, not failed: resending them would fail the same way.

Each request is compressed with zstd and kept under Snowflake's 4 MB request limit. Larger batches improve throughput and cost; aim for batches of at least a few megabytes.

Examples​

Stream JSON to a table​

Streams generated JSON rows to a table, with the key taken from an environment variable.

output:
  snowflake_stream:
    account: myorg-myaccount
    user: INGEST_USER
    role: INGEST_ROLE
    database: ANALYTICS
    schema: PUBLIC
    table: EVENTS
    private_key: ${SNOWFLAKE_PRIVATE_KEY}
    batching:
      count: 10000
      period: 5s

Fields​

account​

The Snowflake account identifier, preferably in organisation-account form (myorg-myaccount). An account locator (xy12345, or xy12345.us-east-2.aws) is also accepted.

Type: string

user​

The Snowflake user to authenticate as.

Type: string

role​

The role to use. When unset, the user's default role applies. The role needs privileges to insert into the target table (or operate the pipe).

Type: string

database​

The target database.

Type: string

schema​

The target schema.

Type: string

table​

The target table. Supports environment-variable interpolation at startup, e.g. events_${NETWORK_ID}.

Type: string

pipe​

The pipe to stream through. When unset, the table's default streaming pipe (<TABLE>-STREAMING) is used, which Snowflake provides without a CREATE PIPE. Set this to use a pipe you created yourself, for example one that transforms rows.

Type: string

private_key​

The user's RSA private key as PEM (PKCS#8, encrypted PKCS#8 or PKCS#1). Newlines collapsed to \n and bare base64 DER are also accepted, so the key can come from an environment variable, e.g. ${SNOWFLAKE_PRIVATE_KEY}. The matching public key must be registered with ALTER USER ... SET RSA_PUBLIC_KEY.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

private_key_file​

Path to a file holding the RSA private key, as an alternative to private_key.

Type: string

private_key_pass​

Passphrase for an encrypted private key.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

url​

Override the account URL, e.g. for private connectivity. Defaults to https://<account>.snowflakecomputing.com. Must use https (plain http is accepted only for a loopback host, such as a local proxy).

Type: string

mapping​

An optional mapping applied to each message before it is sent. The result must be a JSON object whose keys are column names.

Type: string

channel_prefix​

Prefix for the channel names this output opens, one per in-flight batch. The edge appends its node ID, an owner and a short hash of the output's label (or, without one, its place in the config), so every node and every output writes its own channels. The owner is the job when offset_token is set, so those channels (and the committed offset they keep) survive restarts and job updates; otherwise it is the execution, so each job update starts fresh channels. Changing the label or this prefix starts new channels. Defaults to expanso.

Type: string

offset_token​

An offset token evaluated per message, for exactly-once delivery. Messages whose token is at or below the channel's last committed token are skipped, so replays after a restart are not duplicated. Tokens are compared as strings and must increase across every message the channel receives, not just within a batch, so the source must be a single ordered stream (for example one Kafka partition per pipeline). Zero-pad numeric offsets, e.g. ${! "%020d".format(@kafka_offset) }: bare numbers of different lengths are refused. Requires a single channel (max_in_flight: 1); a failed batch, including one with a message whose mapping or token fails to evaluate, is retried before any later one. The committed offset lives on the channel, so the guarantee holds while the channel name does: set a label on the output so it keeps its channel when the config is rearranged (see channel_prefix). Snowflake deletes a channel, and its committed offset, after 30 days without activity. Rows can still repeat in two narrow cases: an append that commits only after its channel was reopened, and the brief overlap of the old and new execution during a job update, which share the channel.

This field supports interpolation functions.

Type: string

max_in_flight​

The maximum number of batches in flight, each on its own channel. Defaults to 4, or 1 when offset_token is set. Snowflake allows 2,000 active channels per pipe by default, so across a fleet, nodes x max_in_flight (for every output on the same table) must stay below that; lower it, or split tables, for large fleets. Without offset_token the channels are dropped when the pipeline stops.

Type: int

commit_backoff​

How the output polls for a batch to be committed before acknowledging it.

Type: object

commit_backoff.initial_interval​

The wait before the first commit-status poll after a batch is sent.

Type: string
Default: "250ms"

commit_backoff.max_interval​

The longest delay between commit-status polls.

Type: string
Default: "512ms"

commit_backoff.max_elapsed_time​

How long to wait for a batch to be committed before failing it for retry.

Type: string
Default: "60s"

commit_backoff.multiplier​

The growth factor between successive poll delays.

Type: float
Default: 2

request_timeout​

How long one request to Snowflake may take, including uploading an append of up to max_request_bytes. On slow links, allow at least max_request_bytes divided by the upload rate, or lower max_request_bytes.

Type: string
Default: "60s"

max_request_bytes​

The largest append request, in compressed bytes (65536 to 4000000, Snowflake's limit). Batches are split into requests of this size; smaller requests suit slow or metered links, larger ones cost fewer requests.

Type: int
Default: 3934464

batching​

How messages are grouped into batches. Each batch waits for its own commit, so without batching every message is sent and committed on its own; set a count and a period.

Type: object

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.check​

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

Type: string
Default: ""

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. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array of processor

_channel_scope​

Internal, filled in by the Expanso runtime. Do not set it; submissions that do are rejected.

Type: string
Default: ""