Working with the Kafka Connector

To understand Kafka’s core concepts and how it works, please read the Kafka documentation. This guide assumes that you understand Kafka’s basic concepts and terminology, and that you have a working Kafka environment up and running.

The Confluent Kafka Connector is available via the Confluent Hub and as a download from SingleStore Helios.

Note: After you have installed the version you want to use, you will need to configure the connector properties.

The rest of this page describes how the connector works.

Note: You can also use a pipeline to Load Data from Kafka Using a Pipeline.

Connector Behavior

See the SingleStore Kafka Connector for information about the connector.

Auto-creation of tables

While loading data, if the table does not exist in SingleStore Helios, it will be created using the information from the first record.

The table name is the name of the topic. The table schema is taken from the record’s valueSchema. If valueSchema is not a struct, then a single column with name data will be created with the schema of the record. Table keys are taken from the tableKey property.

If the table already exists, all records will be loaded directly into it. Automatic schema changes are not supported, so all records should have the same schema.

Exactly once delivery

To achieve exactly once delivery, set singlestore.metadata.allow to true. The kafka_connect_transaction_metadata table will then be created.

This table contains an identifier, count of records, and time of each transaction. The identifier consists of kafka-topic, kafka-partition, and kafka-offset. This combination provides a unique identifier that prevents duplication of data in the SingleStore Helios database. Kafka saves offsets and increases them only if the kafka-connect job succeeds. If the job fails, Kafka will restart the job with the same offset. This means that if the data is written to the database, but the operation fails, Kafka will try to write data with the same offset and metadata identifier to prevent duplication of existing data and simply complete the work successfully.

Data is written to the table and to the kafka_connect_transaction_metadata table in one transaction. Because of this, if an error occurs, no data is added to the database.

To overwrite the name of this table, use the singlestore.metadata.table property.

Data Types

The connector converts Kafka data types to SingleStore Helios data types:

Kafka Type

SingleStore Helios Type

STRUCT

JSON

MAP

JSON

ARRAY

JSON

INT8

TINYINT

INT16

SMALLINT

INT32

INT

INT64

BIGINT

FLOAT32

FLOAT

FLOAT64

DOUBLE

BOOLEAN

TINYINT

BYTES

TEXT

STRING

VARBINARY(1024)

Table Keys

To add a column as a key in SingleStore Helios, use the tableKey property:

Suppose you have an entity:

{
    "id" : 123,
    "name" : "Alice"
}

If you want to add the id column as a PRIMARY KEY to your SingleStore Helios table, add "tableKey.primary": "id" to your properties configuration.

Doing so will generate the following query during table creation:

CREATE TABLE IF NOT EXISTS `table` (
`id` INT NOT NULL,
`name` TEXT NOT NULL,
PRIMARY KEY (`id`)
)

You can also specify the name of a key by providing it like this: "tableKey.primary.someName" : "id".

This will create a key with a name:

CREATE TABLE IF NOT EXISTS `table` (
`id` INT NOT NULL,
`name` TEXT NOT NULL,
PRIMARY KEY `someName`(`id`)
)

Table Names

By default, the Kafka Connector maps data from topics into SingleStore Helios tables by matching the topic name to the table name. For example, if the Kafka topic is called kafka-example-topic then the connector will load it into the SingleStore Helios table called kafka-example-topic. The table will be created if it does not already exist.

To specify a custom table name, you can use the singlestore.tableName.<topicName> property.

{
    ...
    "singlestore.tableName.foo" : "bar",
    ...
}

In this example, data from the Kafka topic foo will be written to the SingleStore Helios table called bar.

You can use this method to specify custom table names for multiple topics:

{
    ...
    "singlestore.tableName.kafka-example-topic-1" : "singlestore-table-name-1",
    "singlestore.tableName.kafka-example-topic-2" : "singlestore-table-name-2",
    ...
}

Last modified: August 16, 2023

Was this article helpful?