Snowflake Kafka Connector: Configure, Map Topics, and Shape Tables

1.Where the Kafka connector runs

Apache Kafka uses a publish/subscribe model. Producers write messages to topics, and consumers read them asynchronously, so a subscriber never has to be connected directly to the publisher. To scale, topics are split into partitions. Kafka Connect is the framework that links Kafka to external systems such as databases. A Kafka Connect cluster is separate from the Kafka cluster itself, and it is where connectors run and scale out.

The Snowflake Connector for Kafka is one of those connectors. It reads one or more topics and loads the records into Snowflake tables. The typical pattern is one topic feeding one table, with each Kafka message becoming one row. Snowflake ships a build for the Confluent package and a build for open-source Apache Kafka. Confluent Cloud also offers a hosted version, so in that case you don't run the Connect workers yourself.

The classic connector loads data with one of two methods: Snowpipe or Snowpipe Streaming. You also need to know which generation of the connector you are using. The classic connector (v3 and earlier) is still fully supported, but Snowflake's documentation says it is planned for future deprecation. A formal announcement was planned for mid-2026, followed by an 18-month migration window. Snowflake recommends v4 for all new implementations. The two generations handle tables differently, so the rest of this page labels each behaviour with its version.

Classic connector buffering. The classic connector buffers records in memory per Kafka partition and flushes them to an internal stage as data files when a threshold is reached. The classic install documentation lists three thresholds: buffer.count.records (default 10000 records), buffer.flush.time (default 120 seconds between flushes) and buffer.size.bytes (default 5000000, or 5 MB). Records are compressed when they are written to data files, so the size of the records in the buffer can be larger than the size of the files created from them.

Sources12

2.Configuring the connection

Connector configuration is a set of properties. Some tell the connector where to write in Snowflake. Others tell Kafka Connect how to deserialize records. The Snowflake-side properties identify the account, the user, the role used for inserts, and the database and schema that hold the target tables. The URL must include your account identifier, and the https:// protocol and port number are optional.

The converter properties belong to Kafka Connect. For JSON records, the value converter is the community JsonConverter. For Avro records with a Schema Registry, it is Confluent's AvroConverter. One restriction to remember: when snowflake.enable.schematization is true (the default), StringConverter and ByteArrayConverter aren't supported as value converters.

Sources3

3.From topic name to table name

After the connector is connected, it has to decide which table each topic writes to. There are two modes. In static mapping, the table name is derived from the topic name alone. In explicit mapping, you list topic-to-table pairs in snowflake.topic2table.map. Without that parameter, the connector always derives the name from the topic.

The classic install documentation describes the map format. Each topic and its table are separated by a colon, and pairs are comma-separated. Several topics can point at one table, for example topic1:low_range,topic2:low_range. Regular expressions are allowed to define topics, as with topics.regex, so topic[0-4]:low_range,topic[5-9]:high_range is valid. The expressions cannot be ambiguous: any matched topic must match only a single target table.

The derivation rules are mechanical. A topic name that is a valid Snowflake identifier becomes the table name in uppercase. Invalid characters are replaced with underscores, and a name that starts with something other than a letter or underscore gets an underscore prepended. Replacing characters can make two different topics produce the same name, so whenever the connector adjusts a name it appends an underscore and a hash code. The v4 docs give the example that topic my-topic.data becomes MY_TOPIC_DATA_<hash>. Snowflake's advice is to choose topic names that are already valid identifiers so table names stay predictable.

Sources42

4.What lands in the table: classic columns versus v4 pipes

Classic connector. By default, every table it loads has two VARIANT columns. RECORD_CONTENT holds the Kafka message (JSON or Avro), and RECORD_METADATA holds details such as the topic, partition and offset. The message is not parsed or split into columns. If Snowflake creates the table, it has only those two columns. If you create the table yourself, you can add more columns, but they must allow NULL because the connector never supplies values for them. You read the data with VARIANT path syntax:

select
       record_metadata:CreateTime,
       record_content:ID
    from table1
    where record_metadata:topic = 'PressureOverloadWarning';

v4 connector. You don't have to create a table or pipe in advance. The connector creates a table whose columns match the JSON keys and uses a default pipe named {tableName}-STREAMING. That pipe maps first-level keys to columns by name, ignoring case. Nested objects and arrays should go into VARIANT columns. Keys that don't match a column are ignored unless ENABLE_SCHEMA_EVOLUTION is enabled on the table, in which case the connector adds the new columns.

If you need renamed columns, casts, masking or filtering, use user-defined pipe mode. You create a pipe with the same name as the destination table and put the transformation in its COPY INTO, reading from the streaming data source. When choosing a pipe, the connector looks for a pipe with the table's name and falls back to the default if there isn't one. User-defined pipes are not available with client-side validation (snowflake.validation=client_side).

CREATE PIPE ORDERS AS
COPY INTO ORDERS
FROM (
  SELECT
  $1:order_id::STRING,
  $1:customer_name,
  $1:order_total::STRING,
  $1:isPaid::STRING
FROM TABLE(DATA_SOURCE(TYPE => ' ? '))
);

Sources4

5.Controlling RECORD_METADATA

In both generations, the connector records where each row came from. RECORD_METADATA carries the topic, the Kafka partition (not a Snowflake micro-partition), the offset, and CreateTime or LogAppendTime in milliseconds since the epoch. It can also carry the message key and headers. With Snowpipe Streaming it includes SnowflakeConnectorPushTime, the moment the record entered the ingestion buffer, which is useful for measuring end-to-end latency. In a v4 user-defined pipe, you can pull individual fields out with $1:RECORD_METADATA.

The key is only stored if key.converter is set to org.apache.kafka.connect.storage.StringConverter. With any other converter, the connector ignores keys. Metadata also has a cost: about 150 bytes per record, depending on the length of the topic name. That can be a large share of a small payload and it affects billing. Boolean flags control which fields are included.

Sources4

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.The classic Kafka connector parses each JSON message into separate typed columns.Why is that wrong?

   By default, the classic connector stores the whole message in a single VARIANT column (RECORD_CONTENT) without parsing it. Column-per-key mapping is v4 default-pipe behaviour.

   Covered in What lands in the table: classic columns versus v4 pipes

2. 2.Message keys always appear in RECORD_METADATA, whatever key.converter is set to.Why is that wrong?

   Keys are stored only when key.converter is set to org.apache.kafka.connect.storage.StringConverter. Otherwise they are ignored.

   Covered in Controlling RECORD_METADATA

3. 3.Columns you add to a classic connector table can be NOT NULL, because the connector fills them.Why is that wrong?

   The connector only writes its own columns. Any extra columns must allow NULL.

   Covered in What lands in the table: classic columns versus v4 pipes