Skip to main content

snowflake_streaming

Allows Snowflake to ingest data from your data pipeline using Snowpipe Streaming.

This output streams messages into a Snowflake table using the Snowpipe Streaming API. It authenticates with an RSA key pair, batches messages before writing, and can optionally evolve the target table's schema as new fields appear in your data.

# Config fields, showing default values
output:
  label: ""
  snowflake_streaming:
    account: "" # No default (required)
    batching:
      byte_size: 0
      check: ""
      count: 0
      period: ""
      processors: []
    build_options:
      chunk_size: 50000
      parallelism: 1
    channel_name: ""
    channel_prefix: ""
    commit_backoff:
      initial_interval: 32ms
      max_elapsed_time: 60s
      max_interval: 512ms
      multiplier: 2
    commit_timeout: ""
    database: "" # No default (required)
    init_statement: ""
    mapping: ""
    max_in_flight: 4
    message_format: object
    offset_token: ""
    parallelism: 0
    private_key: ""
    private_key_file: ""
    private_key_pass: ""
    role: "" # No default (required)
    schema: "" # No default (required)
    schema_evolution:
      enabled: false # No default (required)
      ignore_nulls: true
      new_column_type_mapping: ""
      processors: []
    table: "" # No default (required)
    timestamp_format: "2006-01-02T15:04:05.999999999Z07:00"
    url: ""
    user: "" # No default (required)

Examples​

Stream change events into a Snowflake table with exactly-once delivery by using the source log sequence number as the offset token.

input:
  postgres_cdc:
    dsn: postgres://foouser:foopass@localhost:5432/foodb
    schema: "public"
    tables: ["my_pg_table"]
    batching:
      count: 50000
      period: 45s
    checkpoint_limit: 1
output:
  snowflake_streaming:
    offset_token: "${!@lsn}"
    max_in_flight: 1
    account: "MYSNOW-ACCOUNT"
    user: MYUSER
    role: ACCOUNTADMIN
    database: "MYDATABASE"
    schema: "PUBLIC"
    table: "MY_PG_TABLE"
    private_key_file: "my/private/key.p8"

Fields​

account​

The Snowflake account name to use, formatted as <orgname>-<account_name>.

Type: string

batching​

Allows you to configure a batching policy for messages before they are written to Snowflake.

Type: object

batching.byte_size​

The number of bytes at which an incomplete batch is flushed. Set to 0 to disable size based batching.

Type: int
Default: 0

batching.check​

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

Type: string
Default: ""

batching.count​

The number of messages after which an incomplete batch is flushed. Set to 0 to disable count based batching.

Type: int
Default: 0

batching.period​

The period of time after which an incomplete batch is flushed regardless of its size.

Type: string
Default: ""

batching.processors​

A list of processors to apply to a batch as it is flushed.

Type: array

build_options​

Options that control how batches of rows are built before they are submitted to Snowflake.

Type: object

build_options.chunk_size​

The number of table rows to submit in each chunk when building the output.

Type: int
Default: 50000

build_options.parallelism​

The maximum amount of parallel processing to use when building the output.

Type: int
Default: 1

channel_name​

The channel name to use when connecting to a Snowflake table. This field supports interpolation functions. It cannot be used together with channel_prefix.

Type: string

channel_prefix​

The prefix to use when creating a channel name. Adding a channel_prefix avoids the creation of duplicate channel names. It cannot be used together with channel_name.

Type: string

commit_backoff​

Controls the backoff behavior used when polling Snowflake for the commit status of submitted data.

Type: object

commit_backoff.initial_interval​

The initial period to wait between status polls.

Type: string
Default: "32ms"

commit_backoff.max_elapsed_time​

The maximum total amount of time to wait for data to be committed. If set to zero then no limit is used.

Type: string
Default: "60s"

commit_backoff.max_interval​

The maximum period to wait between status polls.

Type: string
Default: "512ms"

commit_backoff.multiplier​

The factor by which the wait period between status polls grows on each attempt.

Type: float
Default: 2

commit_timeout​

The maximum amount of time to wait for a commit to complete.

Type: string

database​

The Snowflake database to write data to.

Type: string

init_statement​

Optional SQL statements to execute immediately after the output connects to Snowflake. This is useful for creating tables or setting up resources, and should be idempotent as it runs on every restart.

Type: string

mapping​

An optional Bloblang mapping to execute on each message before it is written.

Type: string

max_in_flight​

The maximum number of messages to have in flight at a given time. Increase this value to improve throughput until performance plateaus.

Type: int
Default: 4

message_format​

The format of each message. When set to object, each message is a JSON or Bloblang object whose keys are column names. When set to array, each message is an array of values where each position matches the column ordinal.

Type: string
Default: "object"

offset_token​

The offset token to use for exactly-once delivery of data. This field supports interpolation functions. Messages are dropped if their token is lexicographically less than the latest processed token.

Type: string

parallelism​

The amount of parallel processing to use when writing data to Snowflake.

Type: int

private_key​

The PEM encoded private RSA key to use for authentication with Snowflake. Either this field or private_key_file must be specified. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

Type: string

private_key_file​

A .p8, PEM encoded file to load the private RSA key from. Either this field or private_key must be specified.

Type: string

private_key_pass​

The passphrase for the private RSA key, required when the key is encrypted. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

Type: string

role​

The role of the user specified in the user field, for example ACCOUNTADMIN. The role must have the privileges required to call the Snowpipe Streaming APIs.

Type: string

schema​

The schema of the Snowflake database to write data to.

Type: string

schema_evolution​

Options that control automatic evolution of the target table's schema as new fields appear in your data.

Type: object

schema_evolution.enabled​

Whether schema evolution is enabled. When set to true, the Snowflake table is created automatically based on the schema of the first message written to it, if the table does not already exist, and new columns are added as new fields appear in the data.

Type: bool

schema_evolution.ignore_nulls​

When set to true and schema evolution is enabled, new columns that only have null values are not added to the table. This avoids unnecessary schema changes and prevents columns from being created with an incorrect data type.

Type: bool
Default: true

schema_evolution.new_column_type_mapping​

An optional mapping used to determine the Snowflake column type assigned to new columns that are created during schema evolution.

Type: string

schema_evolution.processors​

A series of processors to execute when new columns are added to the Snowflake table.

Type: array

table​

The Snowflake table to write data to. This field supports interpolation functions.

Type: string

timestamp_format​

The format used to parse string values for TIMESTAMP columns. This should be a layout compatible with Go's time.Parse.

Type: string
Default: "2006-01-02T15:04:05.999999999Z07:00"

url​

An optional custom URL to use when connecting to Snowflake. This overrides the default URL, which is generated automatically from the account name.

Type: string

user​

The user used to run the Snowpipe Streaming session.

Type: string