Working with the Snowflake High Performance connector for Kafka¶

Topic names, table names, and pipe names¶

Depending on the configuration settings, the connector will use different names for the destination table. The destination table name is always derived from the topic name.

Understanding RECORD_METADATA¶

The connector populates the RECORD_METADATA structure with metadata about the Kafka record. This metadata is sent through the Snowpipe Streaming data source to Snowflake, where it becomes available in pipe transformations using the $1:RECORD_METADATA accessor. RECORD_METADATA structure is available in both user-defined pipe and default pipe modes. Its content can be saved to the column of type VARIANT, or individual fields can be extracted and saved to separate columns.

Example pipe with transformations and metadata:

CREATE PIPE ORDERS AS
COPY INTO ORDERS_TABLE
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total,
    $1:RECORD_METADATA.topic AS source_topic,
    $1:RECORD_METADATA.offset::NUMBER AS kafka_offset,
    $1:RECORD_METADATA.SnowflakeConnectorPushTime::BIGINT AS ingestion_time
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

In this example:

• The pipe extracts specific fields from the Kafka message (order_id, customer_name, order_total)

• It also captures metadata fields (topic, offset, and ingestion timestamp)

• The values can be cast and/or transformed as needed

How metadata fields are populated¶

The connector automatically populates metadata fields based on the Kafka record properties and connector configuration. You can control which metadata fields are included using these configuration parameters:

• snowflake.metadata.topic (default: true) - Includes the topic name

• snowflake.metadata.offset.and.partition (default: true) - Includes offset and partition

• snowflake.metadata.createtime (default: true) - Includes the Kafka record timestamp

• snowflake.metadata.all (default: true) - Includes all available metadata

When snowflake.metadata.all=true (the default), all metadata fields are populated. Setting individual metadata flags to false excludes those specific fields from the RECORD_METADATA structure.

The RECORD_METADATA structure contains the following information by default:

The amount of metadata recorded in the RECORD_METADATA column is configurable using optional Kafka configuration properties.

The field names and values are case-sensitive.

How Kafka records are converted before ingestion¶

Before each row is handed over to Snowpipe Streaming, the connector converts the Kafka Connect record value into a Map<String, Object> whose keys must match your target column names (or can be transformed inside a user-defined pipe). Primitive strings, byte arrays, or numbers must be wrapped (for example by using the HoistField SMT) so that the connector receives a structured object. The converter applies the following rules:

• Null values are treated as tombstones. They are skipped when behavior.on.null.values=IGNORE or ingested as empty JSON objects otherwise.

• Numeric and boolean fields are passed through as-is. Decimal values whose precision is greater than 38 are serialized as strings to stay within Snowflake’s NUMBER limits.

• byte[] and ByteBuffer payloads are Base64-encoded strings, so store them in VARIANT or VARCHAR columns.

• Arrays remain arrays, and nested objects remain nested maps. Declare VARIANT columns when you rely on the default pipe to land nested data as-is.

• Maps with non-string keys are emitted as arrays of [key, value] pairs because Snowflake column names must be text.

• Record headers and keys are copied into RECORD_METADATA whenever the relevant metadata flags are enabled.

If you need the entire message body preserved as a single column, wrap it into a new top-level field using SMTs. See Legacy RECORD_CONTENT column for the transformation pattern.

User-defined pipe mode¶

In this mode, you have full control over data transformation and column mapping.

When to use this mode:

• You need custom column names that differ from JSON field names

• You need to apply data transformations (type casting, masking, filtering)

• You want full control over how data is mapped to columns

Default pipe mode¶

In this mode, the connector uses a default pipe named {tableName}-STREAMING and maps kafka record fields to table columns matching by name (case-insensitive).

When to use this mode:

• Your kafka record key names match your desired column names

• You don’t need custom data transformations

• You want a simple configuration

Mapping kafka record keys to table columns with default pipe mode

When using default pipe mode, the connector uses default pipe named {tableName}-STREAMING and maps content’s first-level keys directly to table columns using case-insensitive matching.

Using default pipe mode - example¶

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  "@&$#* includes special characters": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  record_metadata VARIANT,
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  "has cat" BOOLEAN,
  "!@&$#* includes special characters" BOOLEAN,
  skills VARIANT,
  family VARIANT
);

Using user-defined pipe mode - examples¶

This example shows how to configure and use user-defined pipes with custom data transformations.

CREATE TABLE ORDERS (
  order_id NUMBER,
  customer_name VARCHAR,
  order_total NUMBER,
  order_date TIMESTAMP_NTZ,
  source_topic VARCHAR
);

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:order_id::NUMBER,
    $1:customer_name,
    $1:order_total::NUMBER,
    $1:order_date::TIMESTAMP_NTZ,
    $1:RECORD_METADATA.topic
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

snowflake.topic2table.map=kafka-orders-topic:ORDERS

{
  "city": "New York",
  "age": 30,
  "married": true,
  "has cat": true,
  " @&$#* has Łułósżź": true,
  "skills": ["sitting", "standing", "eating"],
  "family": {"son": "Jack", "daughter": "Anna"}
}

CREATE TABLE PERSON_DATA (
  city VARCHAR,
  age NUMBER,
  married BOOLEAN,
  has_cat BOOLEAN,
  weird_field_name BOOLEAN,
  skills VARIANT,
  family VARIANT
);

CREATE PIPE PERSON_DATA AS
COPY INTO PERSON_DATA
FROM (
  SELECT
    $1:city,
    $1:age,
    $1:married,
    $1['has cat'] AS has_cat,
    $1[' @&$#* has Łułósżź'] AS weird_field_name,
    $1:skills,
    $1:family
  FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Explicit topic-to-table mapping¶

When you configure the snowflake.topic2table.map parameter, the connector operates in explicit mapping mode. This mode allows you to:

• Map multiple Kafka topics to a single Snowflake table

• Use custom table names that differ from topic names

• Apply regex patterns to match multiple topics

Configuration format:

The snowflake.topic2table.map parameter accepts a comma-separated list of topic-to-table mappings in the format:

topic1:table1,topic2:table2,topic3:table3

Example configurations:

Direct topic mapping

snowflake.topic2table.map=orders:ORDER_TABLE,customers:CUSTOMER_TABLE

Regex pattern matching

snowflake.topic2table.map=.*_cat:CAT_TABLE,.*_dog:DOG_TABLE

This configuration maps all topics ending with _cat (such as orange_cat, calico_cat) to the CAT_TABLE table, and all topics ending with _dog to the DOG_TABLE table.

Many topics to one table

snowflake.topic2table.map=topic1:shared_table,topic2:shared_table,topic3:other_table

This configuration maps both topic1 and topic2 to shared_table, while topic3 maps to other_table.

Legacy RECORD_CONTENT column¶

In prior versions of the connector (3.x and earlier), when the schematization feature was disabled, the connector created a destination table with two columns: RECORD_CONTENT and RECORD_METADATA. The RECORD_CONTENT column contained the entire Kafka message content in a column of type VARIANT. The RECORD_METADATA column continues to be supported but the RECORD_CONTENT column is no longer created by the connector. The same functionality can be achieved using SMT transformations (see examples later in this section). The RECORD_CONTENT key is also no longer available in PIPE transformations. For example, this PIPE definition will not work by default:

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:RECORD_CONTENT
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

If you need entire Kafka message content saved to a single column, or you need a handle to the entire Kafka message content in a PIPE transformation, you can use the following SMT transformation that wraps the entire Kafka message content into your desired custom field:

transforms=wrapKafkaMessageContent
transforms.wrapKafkaMessageContent.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.wrapKafkaMessageContent.field=your_top_level_field_name

This transformation will wrap the entire Kafka message content into a custom field named your_top_level_field_name. You can then access the entire Kafka message content using the $1:your_top_level_field_name accessor in your PIPE transformation.

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
    $1:your_top_level_field_name
FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

Alternatively, if you want to save both the entire metadata and content to a single table using the default pipe, do not create a custom pipe; instead, create only a table with two columns: RECORD_CONTENT and your_top_level_field_name.

CREATE TABLE ORDERS (
  record_metadata VARIANT,
  your_top_level_field_name VARIANT
);

To read more about the HoistField$Value transformation, see the Kafka documentation.

Handling streaming channel errors and dead-letter queues¶

The connector inspects the Snowpipe Streaming channel status before committing offsets in Kafka. If the connector detects that the rowsErrorCount property on channel has increased since the connector was started, it raises a fatal error (ERROR_5030) when errors.tolerance=none so that data issues don’t go unnoticed. To allow ingestion to continue while triaging bad rows, set errors.tolerance=all

errors.tolerance=all

Grant usage on an external volume¶

To grant USAGE privilege on the external volume associated with your Apache Iceberg™ table to your role for the Kafka connector, run the following statement:

For example, if your Iceberg table uses the kafka_external_volume external volume and the connector uses the role kafka_connector_role, run the following statement:

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kafka_external_volume TO ROLE kafka_connector_role;

Create an Apache Iceberg™ table for ingestion¶

The connector does not create Iceberg tables automatically and does not support schema evolution. Before you run the connector, you must create an Iceberg table manually.

When you create an Iceberg table, you can use Iceberg data types (including VARIANT) or compatible Snowflake types.

For example, consider the following message:

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "options":
    {
        "can_walk": true,
        "can_talk": false
    },
    "date_added": "2024-10-15"
}

To create an Iceberg table for the example message, use one of the following statements:

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id number(38,0),
    name varchar,
    body_temperature number(4,2),
    approved_coffee_types array(varchar),
    animals_possessed variant,
    options object(can_walk boolean, can_talk boolean),
    date_added date
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    id INT,
    name string,
    body_temperature float,
    approved_coffee_types array(string),
    animals_possessed variant,
    date_added date,
    options object(can_walk boolean, can_talk boolean),
    )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ICEBERG_VERSION = 3;

Copy