Skip to main content

snowflake_streaming

Ingest data into Snowflake using Snowpipe Streaming.

# Common config fields, showing default values
output:
  label: ""
  snowflake_streaming:
    account: "" # No default (required)
    user: "" # No default (required)
    role: "" # No default (required)
    database: "" # No default (required)
    schema: "" # No default (required)
    table: "" # No default (required)
    private_key: "" # No default (optional)
    private_key_file: "" # No default (optional)
    private_key_pass: "" # No default (optional)
    mapping: "" # No default (optional)
    init_statement: "" # No default (optional)
    schema_evolution:
      enabled: false # No default (required)
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
    max_in_flight: 4

Ingest data into Snowflake using Snowpipe Streaming.

[%header,format=dsv] |=== Snowflake column type:Allowed format in Expanso Edge CHAR, VARCHAR:string BINARY:[]byte NUMBER:any numeric type, string FLOAT:any numeric type BOOLEAN:bool,any numeric type,string parsable according to strconv.ParseBool TIME,DATE,TIMESTAMP:unix or RFC 3339 with nanoseconds timestamps VARIANT,ARRAY,OBJECT:any data type is converted into JSON GEOGRAPHY,GEOMETRY: Not supported |===

For TIMESTAMP, TIME and DATE columns, you can parse different string formats using a bloblang mapping.

Authentication can be configured using a RSA Key Pair.

There are limitations of what data types can be loaded into Snowflake using this method.

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.

It is recommended that each batches results in at least 16MiB of compressed output being written to Snowflake. You can monitor the output batch size using the snowflake_compressed_output_size_bytes metric.

Examples​

Exactly once CDC into Snowflake​

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

NOTE: If attempting to do exactly-once it's important that rows are delivered in order to the output. Be sure to read the documentation for offset_token first. Removing the offset_token is a safer option that will instruct Expanso Edge to use its default at-least-once delivery model instead.

input:
  postgres_cdc:
    dsn: postgres://foouser:foopass@localhost:5432/foodb
    schema: "public"
    slot_name: "my_repl_slot"
    tables: ["my_pg_table"]
    # 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
    batching:
      count: 50000
      period: 45s
    # Prevent multiple batches from being in flight at once, so that we never send
    # a batch while another batch is being retried, this is important to ensure that
    # the Snowflake Snowpipe Streaming channel does not see older data - as it will
    # assume that the older data is already committed.
    checkpoint_limit: 1
output:
  snowflake_streaming:
    # We use the log sequence number in the WAL from Postgres to ensure we
    # only upload data exactly once, these are already lexicographically
    # ordered.
    offset_token: "${!@lsn}"
    # Since we're sending a single ordered log, we can only send one thing
    # at a time to ensure that we're properly incrementing our offset_token
    # and only using a single channel at a time.
    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"

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.

NOTE: If attempting to do exactly-once its important that records are delivered in order to the output and correctly partitioned. Be sure to read the documentation for channel_name and offset_token first. Removing the offset_token is a safer option that will instruct Expanso Edge to use its default at-least-once delivery model instead.

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: "${TODO}"
output:
  fallback:
    - snowflake_streaming:
        # To ensure that we write an ordered stream each partition in kafka gets its own
        # channel.
        channel_name: "partition-${!@kafka_partition}"
        # Ensure that our offsets are lexicographically sorted in string form by padding with
        # leading zeros
        offset_token: offset-${!"%016X".format(@kafka_offset)}
        account: "MYSNOW-ACCOUNT"
        user: MYUSER
        role: ACCOUNTADMIN
        database: "MYDATABASE"
        schema: "PUBLIC"
        table: "MYTABLE"
        private_key_file: "my/private/key.p8"
        schema_evolution:
          enabled: true
    # 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.

NOTE: 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
    # Batching policy, influences how large the generated files sent to Snowflake are
    batch_policy:
      enabled: true
      byte_size: 33554432 # 32MiB
      period: "10s"
output:
  snowflake_streaming:
    account: "MYSNOW-ACCOUNT"
    user: MYUSER
    role: ACCOUNTADMIN
    database: "MYDATABASE"
    schema: "PUBLIC"
    table: "MYTABLE"
    private_key_file: "my/private/key.p8"
    # By default there is only a single channel per output table allowed
    # if we want to have multiple Redpanda Connect streams writing data
    # then we need a unique channel prefix per stream. We'll use the host
    # name to get unique prefixes in this example.
    channel_prefix: "snowflake-channel-for-${HOST}"
    schema_evolution:
      enabled: true

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

url​

Override the default URL used to connect to Snowflake which is https://ORG-ACCOUNT.snowflakecomputing.com

Type: string

user​

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

Type: string

role​

The role for the user field. The role must have the required privileges to call the Snowpipe Streaming APIs. See Snowflake Documentation for more information about roles.

Type: string

database​

The Snowflake database to ingest data into.

Type: string

schema​

The Snowflake schema to ingest data into.

Type: string

table​

The Snowflake table to ingest data into.

This field supports interpolation functions.

Type: string

private_key​

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

Secret

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

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.

Secret

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

Type: string

mapping​

A bloblang mapping to execute on each message.

Type: string

init_statement​

Optional SQL statements to execute immediately upon the first connection. This is a useful way to initialize tables before processing data. Care should be taken to ensure that the statement is idempotent, and therefore would not cause issues when run multiple times after service restarts.

Type: string

schema_evolution​

Options to control schema evolution within the pipeline as new columns are added to the pipeline.

Type: object

schema_evolution.enabled​

Whether schema evolution is enabled.

Type: bool

schema_evolution.ignore_nulls​

If true, then new columns that are null are ignored and schema evolution is not triggered. If false then null columns trigger schema migrations in Snowflake. NOTE: unless you already know what type this column will be in advance, it's highly encouraged to ignore null values.

Type: bool
Default: true

schema_evolution.processors​

A series of processors to execute when new columns are added to the table. Specifying this can support running side effects when the schema evolves or enriching the message with additional data to guide the schema changes. For example, one could read the schema the message was produced with from the schema registry and use that to decide which type the new column in Snowflake should be.

The input to these processors is an object with the value and the name of the new column, the original message and table being written too. The metadata is unchanged from the original message that caused the schema to change. For example: {"value": 42.3, "name":"new_data_field", "message": {"existing_data_field": 42, "new_data_field": "foo"}, "db": MY_DATABASE", "schema": "MY_SCHEMA", "table": "MY_TABLE"}. The output of these series of processors should be a single message, where the contents of the message is a string indicating the column data type to use (FLOAT, VARIANT, NUMBER(38, 0), etc. An ALTER TABLE statement will then be executed on the table in Snowflake to add the column with the corresponding data type.

Type: array of processor

build_options​

Options to optimize the time to build output data that is sent to Snowflake. The metric to watch to see if you need to change this is snowflake_build_output_latency_ns.

Type: object

build_options.parallelism​

The maximum amount of parallelism to use.

Type: int
Default: 1

build_options.chunk_size​

The number of rows to chunk for parallelization.

Type: int
Default: 50000

batching​

Allows you to configure a batching policy.

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

max_in_flight​

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

Type: int
Default: 4

channel_prefix​

The prefix to use when creating a channel name. Duplicate channel names will result in errors and prevent multiple instances of Expanso Edge from writing at the same time. By default if neither channel_prefix or `channel_name is specified then the output will create a channel name that is based on the table FQN so there will only be a single stream per table.

At most max_in_flight channels will be opened.

This option is mutually exclusive with channel_name.

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

Type: string

channel_name​

The channel name to use. Duplicate channel names will result in errors and prevent multiple instances of Expanso Edge from writing at the same time. Note that batches are assumed to all contain messages for the same channel, so this interpolation is only executed on the first message in each batch. It's recommended to batch at the input level to ensure that batches contain messages for the same channel if using an input that is partitioned (such as an Apache Kafka topic).

This option is mutually exclusive with channel_prefix.

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

This field supports interpolation functions.

Type: string

offset_token​

The offset token to use for exactly once delivery of data in the pipeline. When data is sent on a channel, each message in a batch's offset token is compared to the latest token for a channel. If the offset token is lexicographically less than the latest in the channel, it's assumed the message is a duplicate and is dropped. This means it is very important to have ordered delivery to the output, any out of order messages to the output will be seen as duplicates and dropped. Specifically this means that retried messages could be seen as duplicates if later messages have succeeded in the meantime, so in most circumstances a dead letter queue output should be employed for failed messages.

NOTE: It's assumed that messages within a batch are in increasing order by offset token, additionally if you're using a numeric value as an offset token, make sure to pad the value so that it's lexicographically ordered in its string representation, since offset tokens are compared in string form.

For more information about offset tokens, see ^Snowflake Documentation

This field supports interpolation functions.

Type: string

commit_backoff​

Control how frequently Snowflake is polled to check if data has been committed.

Type: object

commit_backoff.initial_interval​

The initial period to wait between status polls.

Type: string
Default: "32ms"

commit_backoff.max_interval​

The maximum period to wait between status polls.

Type: string
Default: "512ms"

commit_backoff.max_elapsed_time​

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

Type: string
Default: "60s"

commit_backoff.multiplier​

The factor by which the poll interval grows on each attempt.

Type: float
Default: 2

message_format​

The format at which to expect incoming messages from the rest of the pipeline in.

Type: string
Default: "object"

OptionSummary
arrayMessages are an array of values where the position in the array matches up the with ordinal of the column in snowflake
objectMessages are an object in JSON or bloblang where the key of the object is the column name in snowflake and the value is the value for the column

timestamp_format​

The format to parse string values for TIMESTAMP, TIMESTAMP_LTZ and TIMESTAMP_NTZ columns. Should be a layout for ^time.Parse in Golang.

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