Load Data from the Confluent Kafka Connector

Auto-creation of tables

While loading data, if the table does not exist in SingleStoreDB, 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 SingleStoreDB 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 SingleStoreDB data types:

Table Keys

To add a column as a key in SingleStoreDB, 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 SingleStoreDB 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 SingleStoreDB 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 SingleStoreDB 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 SingleStoreDB 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",
    ...
}