Kafka connector¶

You can set up a Kafka connector to stream data from a Kafka topic into Tinybird by creating a connection and .datasource file.

Compatibility¶

The connector is compatible with Apache Kafka and works with any compatible implementation and vendor. The following are tried and tested:

Apache Kafka
Confluent Platform and Confluent Cloud
Redpanda
AWS MSK
Azure Event Hubs for Apache Kafka
Estuary

Set up the connector¶

To set up the Kafka connector, follow these steps.

Create a Kafka Connection¶

Create a .connection file with the required credentials stored in secrets. For example:

kafka_sample.connection

TYPE kafka
KAFKA_BOOTSTRAP_SERVERS {{ tb_secret("KAFKA_SERVERS") }}
KAFKA_SECURITY_PROTOCOL SASL_SSL
KAFKA_SASL_MECHANISM PLAIN
KAFKA_KEY {{ tb_secret("KAFKA_USERNAME") }}
KAFKA_SECRET {{ tb_secret("KAFKA_PASSWORD") }}

For a complete list of Kafka connection settings, see Settings.

Set the values of the secrets using tb secret:

tb [--cloud] secret set KAFKA_SERVERS kafka:9092

Secrets are only replaced in your resources when you deploy. If you change a secret, you need to deploy for the changes to take effect.

Create a Kafka data source¶

Create a .datasource file using tb create --prompt or manually.

Define the data source schema as with any non-Kafka datasource and specify the required Kafka settings. The value of KAFKA_CONNECTION_NAME must match the name of the .connection file you created in the previous step.

For example:

kafka_sample.datasource

SCHEMA >
   `timestamp` DateTime(3) `json:$.timestamp`,
   `session_id` String `json:$.session_id`,
   `action` LowCardinality(String) `json:$.action`,
   `version` LowCardinality(String) `json:$.version`,
   `payload` String `json:$.payload`

ENGINE "MergeTree"
ENGINE_PARTITION_KEY "toYYYYMM(timestamp)"
ENGINE_SORTING_KEY "timestamp"

KAFKA_CONNECTION_NAME kafka_sample # The name of the .connection file
KAFKA_TOPIC test_topic
KAFKA_GROUP_ID {{ tb_secret("KAFKA_GROUP_ID") }}

In addition to the columns specified in SCHEMA, Kafka data sources have additional columns that store metadata of the messages ingested. See Kafka meta columns for more information.

For a complete list of Kafka data source settings, see Settings.

Use different consumer groups for different environments to isolate consumers and their committed offset.

Connectivity check¶

After defining your Kafka data source and connection, validate the setup by running a deploy check:

tb --cloud deploy --check

This will check that the Kafka broker is reachable and that Tinybird can connect to it with the provided credentials. Remember to set any secrets used by the connection.

.datasource Settings¶

Instruction	Required	Description
`KAFKA_CONNECTION_NAME`	Yes	Name of the configured Kafka connection in Tinybird. It must match the name of the connection file (without the extension).
`KAFKA_TOPIC`	Yes	Name of the Kafka topic to consume from.
`KAFKA_GROUP_ID`	Yes	Consumer Group ID to use when consuming from Kafka.
`KAFKA_AUTO_OFFSET_RESET`	No	Offset to use when no previous offset can be found, like when creating a new consumer. Supported values are `latest` and `earliest`. Default: `latest`.
`KAFKA_STORE_HEADERS`	No	Adds a `__headers Map(String, String)` column to the data source, and stores Kafka headers in it for later processing. Default value is `False`.
`KAFKA_STORE_RAW_VALUE`	No	Stores the raw message in its entirety in the `__value` column. Default: `False`.
`KAFKA_KEY_FORMAT`	No	Format of the message key. Valid values are `avro`, `json_with_schema`, and `json_without_schema`. Using `avro` or `json_with_schema` requires `KAFKA_SCHEMA_REGISTRY_URL` to be set in the connection file used by the data source.
`KAFKA_VALUE_FORMAT`	No	Format of the message value. Valid values are `avro`, `json_with_schema`, and `json_without_schema`. Using `avro` or `json_with_schema` requires `KAFKA_SCHEMA_REGISTRY_URL` to be set in the connection file used by the data source.

.connection Settings¶

Instruction	Required	Description
`KAFKA_BOOTSTRAP_SERVERS`	Yes	Comma-separated list of one or more Kafka brokers, including Port numbers.
`KAFKA_KEY`	Yes	Key used to authenticate with Kafka. Sometimes called Key, Client Key, or Username depending on the Kafka distribution.
`KAFKA_SECRET`	Yes	Secret used to authenticate with Kafka. Sometimes called Secret, Secret Key, or Password depending on the Kafka distribution.
`KAFKA_SECURITY_PROTOCOL`	No	Security protocol for the connection. Accepted values are `PLAINTEXT` and `SASL_SSL`. Default value is `SASL_SSL`.
`KAFKA_SASL_MECHANISM`	No	SASL mechanism to use for authentication. Supported values are `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`. Default value is `PLAIN`.
`KAFKA_SCHEMA_REGISTRY_URL`	No	URL of the Kafka schema registry. Used for `avro` and `json_with_schema` deserialization of keys and values. If Basic Auth is required, it must be included in the URL as in `https://user:password@registry_url`

Kafka connector in the local environment¶

You can use the Kafka connector in the Tinybird Local container to consume messages from a local Kafka server or a Kafka server in the cloud.

Local Kafka Server¶

When using a local Kafka server, make sure that the Tinybird Local container can access your local Kafka server. If you are running Kafka using Docker, you can use the following command to connect your local Kafka server to the Tinybird Local container:

docker network connect local-kafka-network tinybird-local

Then, create the required secrets in your local environment. For example:

tb --local secret set KAFKA_KEY "kafka-local-username"

You can also use the tb_secret() default value. For example:

kafkasample.connection

KAFKA_KEY {{ tb_secret("KAFKA_KEY", "kafka-local-username") }}

Kafka server in the cloud¶

When using a Kafka server in the cloud that's visible on your network, you can create the required secrets in your local environment. For example:

tb --local secret set KAFKA_KEY "kafka-cloud-username"

Don't use the tb_secret() default value when using a Kafka server in the cloud, as it might expose credentials in your code.

Kafka meta columns¶

When you connect a data source to Kafka, the following columns are added to store metadata from Kafka messages:

name	type	description
`__value`	`String`	A String representing the entire unparsed value of the Kafka message. It is only populated if `KAFKA_STORE_RAW_VALUE` is set to `True`.
`__topic`	`LowCardinality(String)`	The topic that the message was read from.
`__partition`	`Int16`	The partition that the message was read from.
`__offset`	`Int16`	The offset of the message.
`__timestamp`	`Datetime`	The timestamp of the message.
`__key`	`String`	The key of the message.

Optionally, when KAFKA_STORE_HEADERS is set to True, the following column is added and populated:

name	type	description
`__headers`	`Map(String, String)`	Kafka headers of the message.

When you iterate your Kafka data source, you might need to use the meta columns in the FORWARD_QUERY. Tinybird suggests a valid forward query that you can tweak to get the desired values for each column.

Kafka logs¶

You can find global logs in the datasources_ops_log Service Data Source. Filter by datasource_id to select the correct datasource, and by event_type='append-kafka'.

For example, to select all Kafka releated logs in the last day, run the following query:

SELECT *
FROM tinybird.datasources_ops_log
WHERE datasource_id = 't_1234'
  AND event_type = 'append-kafka'
  AND timestamp > now() - INTERVAL 1 day
ORDER BY timestamp DESC

If you can't find logs in datasources_ops_log, the kafka_ops_log Service Data Source contains more detailed logs. Filter by datasource_id to select the correct datasource, and use msg_type to select the desired log level (info, warning, or error).

SELECT *
FROM tinybird.kafka_ops_log
WHERE datasource_id = 't_1234'
  AND timestamp > now() - interval 1 day
  AND msg_type IN ['info', 'warning', 'error']

Troubleshooting¶

Each combination of KAFKA_TOPIC and KAFKA_GROUP_ID can only be used in one data source, otherwise the offsets committed by the consumers of different data sources clash.

If you connect a data source to Kafka using a KAFKA_TOPIC and KAFKA_GROUP_ID that were previously used by another data source in your workspace, the data source only receives data from the last committed offset, even if KAFKA_AUTO_OFFSET_RESET is set to earliest.

To prevent these issues, always use unique KAFKA_GROUP_IDs when testing Kafka data sources.

See Kafka logs to learn how to diagnose any other issues

Compressed messages¶

Tinybird can consume from Kafka topics where Kafka compression is turned on; decompressing the message is a standard function of the Kafka consumer. However, if you compressed the message before passing it through the Kafka producer, Tinybird can't do post-consumer processing to decompress the message.

For example, if you compressed a JSON message through gzip and produced it to a Kafka topic as a bytes message, it would be ingested by Tinybird as bytes. If you produced a JSON message to a Kafka topic with the Kafka producer setting compression.type=gzip, while it would be stored in Kafka as compressed bytes, it would be decoded on ingestion and arrive to Tinybird as JSON.

Connecting an existing data source to Kafka¶

You can connect an existing, default data source to Kafka.

Create the Kafka .connection file if it does not exist, add the desired Kafka settings to the .datasource file, and add a forward query to provide default values for the the Kafka meta columns.

Disconnecting a data source from Kafka¶

To disconnect a data source from Kafka, remove the Kafka settings from the .datasource file.

If you want to keep any of the Kafka meta columns, add them to the schema with a default value and adjust the FORWARD_QUERY accordingly.

Get started

Get data in

Work with data

Publish your data

Test and deploy

Monitor your data

Datafiles

Command reference