Using Snowflake Connector for Kafka With Snowpipe Streaming¶

Required Properties¶

Add or edit your connection settings in the Kafka connector properties file. For more information, see Configuring the Kafka Connector.

snowflake.ingestion.method

Required only if using the Kafka connector as the streaming ingest client. Specifies whether to use Snowpipe Streaming or standard Snowpipe to load your Kafka topic data. The supported values are as follows:

• SNOWPIPE_STREAMING

• SNOWPIPE (default)

No additional settings are required to choose the backend service to queue and load topic data. Configure additional properties in your Kafka connector properties file as usual.

snowflake.role.name: Access control role to use when inserting the rows into the table.

Buffer and Polling Properties¶

buffer.flush.time

Number of seconds between buffer flushes, with each flush resulting in insert operations for the buffered records. The Kafka connector calls the Snowpipe Streaming API (“API”) once after each flush.

The minimum value supported for the buffer.flush.time property is 1 (in seconds). For higher average data flow rates, we suggest that you decrease the default value for improved latency. If cost is a greater concern than latency, you could increase the buffer flush time. Be careful to flush the Kafka memory buffer before it becomes full to avoid out of memory exceptions.

Values

1 - No upper limit.

Default

10

buffer.count.records

Number of records buffered in memory per Kafka partition before ingesting to Snowflake.

Values

1 - No upper limit.

Default

10000

buffer.size.bytes

Cumulative size in bytes of records buffered in memory per the Kafka partition before they are ingested in Snowflake as data files.

The records are compressed when they are written to data files. As a result, the size of the records in the buffer may be larger than the size of the data files created from the records.

Values

1 - No upper limit.

Default

20000000 (20 MB)

In addition to the Kafka connector properties, note the Kafka consumer max.poll.records property, which controls the maximum number of records returned by Kafka to Kafka Connect in a single poll. The default value of 500 can be increased, but be mindful of memory constraints. For more information about this property, see the documentation for your Kafka package:

• Apache Kafka

• Confluent

Error Handling and DLQ Properties¶

errors.tolerance

Specifies how to handle errors encountered by the Kafka connector:

This property supports the following values:

NONE

Stop loading data when the first error is encountered.

ALL

Ignore all errors and continue to load data.

Default

NONE

errors.log.enable

Specifies whether or not to write error messages to the Kafka Connect log file.

This property supports the following values:

TRUE

Write error messages.

FALSE

Do not write error messages.

Default

FALSE

errors.deadletterqueue.topic.name

Specifies the name of the DLQ (dead-letter queue) topic name in Kafka for delivering messages to Kafka that could not be ingested into Snowflake tables. For more information, see Dead-letter Queues (in this topic).

Values

Custom text string

Default

None.