1 of 1

Apache Kafka

This guide shows you how to ingest a stream of records from an Apache Kafka topic into a Pinot table.

In this page, you'll learn how to import data into Pinot using Apache Kafka for real-time stream ingestion. Pinot has out-of-the-box real-time ingestion support for Kafka.

Let's set up a demo Kafka cluster locally, and create a sample topic transcript-topic

Start Kafka

docker run \
    --network pinot-demo --name=kafka

Create a Kafka topic

docker exec \
  -t kafka \

Start Kafka

Start Kafka cluster on port 9092 using the same Zookeeper from the .

Create a Kafka topic

Download the latest . Create a topic.

Create schema configuration

We will publish the data in the same format as mentioned in the docs. So you can use the same schema mentioned under .

Create table configuration

The real-time table configuration for the transcript table described in the schema from the previous step.

For Kafka, we use streamType as kafka . See for available decoder class options. You can also write your own decoder by extending the StreamMessageDecoder interface and putting the jar file in plugins directory.

The lowLevel consumer reads data per partition whereas the highLevel consumer utilises Kafka high level consumer to read data from the whole stream. It doesn't have the control over which partition to read at a particular momemt.

For Kafka versions below 2.X, use org.apache.pinot.plugin.stream.kafka09.KafkaConsumerFactory

For Kafka version 2.X and above, use org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory

You can set the offset to -

smallest to start consumer from the earliest offset
largest to start consumer from the latest offset
timestamp in format yyyy-MM-dd'T'HH:mm:ss.SSSZ

The resulting configuration should look as follows -

Upload schema and table

Now that we have our table and schema configurations, let's upload them to the Pinot cluster. As soon as the real-time table is created, it will begin ingesting available records from the Kafka topic.

Add sample data to the Kafka topic

We will publish data in the following format to Kafka. Let us save the data in a file named as transcript.json.

Push sample JSON into the transcript-topic Kafka topic, using the Kafka console producer. This will add 12 records to the topic described in the transcript.json file.

Checkin Kafka docker container

Publish messages to the target topic

Query the table

As soon as data flows into the stream, the Pinot table will consume it and it will be ready for querying. Head over to the to checkout the real-time data.

Kafka ingestion guidelines

Kafka versions in Pinot

Pinot supports 2 major generations of Kafka library - kafka-0.9 and kafka-2.x for both high and low level consumers.

Post release 0.10.0, we have started shading kafka packages inside Pinot. If you are using our latest tagged docker images or master build, you should replace org.apache.kafka with shaded.org.apache.kafka in your table config.

Upgrade from Kafka 0.9 connector to Kafka 2.x connector

Update table config for both high level and low level consumer: Update config: stream.kafka.consumer.factory.class.name from org.apache.pinot.core.realtime.impl.kafka.KafkaConsumerFactory to org.apache.pinot.core.realtime.impl.kafka2.KafkaConsumerFactory.
If using Stream(High) level consumer, also add config stream.kafka.hlc.bootstrap.server into tableIndexConfig.streamConfigs. This config should be the URI of Kafka broker lists, e.g.

How to consume from a Kafka version > 2.0.0

This connector is also suitable for Kafka lib version higher than 2.0.0. In , change the kafka.lib.version from 2.0.0 to 2.1.1 will make this Connector working with Kafka 2.1.1.

Kafka configurations in Pinot

Use Kafka partition (low) level consumer with SSL

Here is an example config which uses SSL based authentication to talk with kafka and schema-registry. Notice there are two sets of SSL options, ones starting with ssl. are for kafka consumer and ones with stream.kafka.decoder.prop.schema.registry. are for SchemaRegistryClient used by KafkaConfluentSchemaRegistryAvroMessageDecoder.

Consume transactionally-committed messages

The connector with Kafka library 2.0+ supports Kafka transactions. The transaction support is controlled by config kafka.isolation.level in Kafka stream config, which can be read_committed or read_uncommitted (default). Setting it to read_committed will ingest transactionally committed messages in Kafka stream only.

For example,

Note that the default value of this config read_uncommitted to read all messages. Also, this config supports low-level consumer only.

Use Kafka partition (low) level consumer with SASL_SSL

Here is an example config which uses SASL_SSL based authentication to talk with kafka and schema-registry. Notice there are two sets of SSL options, some for kafka consumer and ones with stream.kafka.decoder.prop.schema.registry. are for SchemaRegistryClient used by KafkaConfluentSchemaRegistryAvroMessageDecoder.

Extract record headers as Pinot table columns

Pinot's Kafka connector supports automatically extracting record headers and metadata into the Pinot table columns. The following table shows the mapping for record header/metadata to Pinot table column names:

Kafka Record

Pinot Table Column

Description

In order to enable the metadata extraction in a Kafka table, you can set the stream config metadata.populate to true.

In addition to this, if you want to use any of these columns in your table, you have to list them explicitly in your table's schema.

For example, if you want to add only the offset and key as dimension columns in your Pinot table, it can listed in the schema as follows:

Once the schema is updated, these columns are similar to any other pinot column. You can apply ingestion transforms and / or define indexes on them.

Remember to follow the when updating schema of an existing table!

Tell Pinot where to find an Avro schema

There is a standalone utility to generate the schema from an Avro file. See [infer the pinot schema from the avro schema and JSON data]() for details.

To avoid errors like The Avro schema must be provided, designate the location of the schema in your streamConfigs section. For example, if your current section contains the following:

Then add this key: "stream.kafka.decoder.prop.schema"followed by a value that denotes the location of your schema.

Apache Kafka

This guide shows you how to ingest a stream of records from an Apache Kafka topic into a Pinot table.

In this page, you'll learn how to import data into Pinot using Apache Kafka for real-time stream ingestion. Pinot has out-of-the-box real-time ingestion support for Kafka.

Let's set up a demo Kafka cluster locally, and create a sample topic transcript-topic

Start Kafka

docker run \
    --network pinot-demo --name=kafka

Create a Kafka topic

docker exec \
  -t kafka \

Start Kafka

Start Kafka cluster on port 9092 using the same Zookeeper from the .

Create a Kafka topic

Download the latest . Create a topic.

Create schema configuration

We will publish the data in the same format as mentioned in the docs. So you can use the same schema mentioned under .

Create table configuration

The real-time table configuration for the transcript table described in the schema from the previous step.

For Kafka versions below 2.X, use org.apache.pinot.plugin.stream.kafka09.KafkaConsumerFactory

For Kafka version 2.X and above, use org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory

You can set the offset to -

smallest to start consumer from the earliest offset
largest to start consumer from the latest offset
timestamp in format yyyy-MM-dd'T'HH:mm:ss.SSSZ

The resulting configuration should look as follows -

Upload schema and table

Add sample data to the Kafka topic

We will publish data in the following format to Kafka. Let us save the data in a file named as transcript.json.

Push sample JSON into the transcript-topic Kafka topic, using the Kafka console producer. This will add 12 records to the topic described in the transcript.json file.

Checkin Kafka docker container

Publish messages to the target topic

Query the table

As soon as data flows into the stream, the Pinot table will consume it and it will be ready for querying. Head over to the to checkout the real-time data.

Kafka ingestion guidelines

Kafka versions in Pinot

Pinot supports 2 major generations of Kafka library - kafka-0.9 and kafka-2.x for both high and low level consumers.

Upgrade from Kafka 0.9 connector to Kafka 2.x connector

Update table config for both high level and low level consumer: Update config: stream.kafka.consumer.factory.class.name from org.apache.pinot.core.realtime.impl.kafka.KafkaConsumerFactory to org.apache.pinot.core.realtime.impl.kafka2.KafkaConsumerFactory.
If using Stream(High) level consumer, also add config stream.kafka.hlc.bootstrap.server into tableIndexConfig.streamConfigs. This config should be the URI of Kafka broker lists, e.g.

How to consume from a Kafka version > 2.0.0

This connector is also suitable for Kafka lib version higher than 2.0.0. In , change the kafka.lib.version from 2.0.0 to 2.1.1 will make this Connector working with Kafka 2.1.1.

Kafka configurations in Pinot

Use Kafka partition (low) level consumer with SSL

Consume transactionally-committed messages

For example,

Note that the default value of this config read_uncommitted to read all messages. Also, this config supports low-level consumer only.

Use Kafka partition (low) level consumer with SASL_SSL

Extract record headers as Pinot table columns

Kafka Record

Pinot Table Column

Description

In order to enable the metadata extraction in a Kafka table, you can set the stream config metadata.populate to true.

In addition to this, if you want to use any of these columns in your table, you have to list them explicitly in your table's schema.

For example, if you want to add only the offset and key as dimension columns in your Pinot table, it can listed in the schema as follows:

Once the schema is updated, these columns are similar to any other pinot column. You can apply ingestion transforms and / or define indexes on them.

Remember to follow the when updating schema of an existing table!

Tell Pinot where to find an Avro schema

There is a standalone utility to generate the schema from an Avro file. See [infer the pinot schema from the avro schema and JSON data]() for details.

To avoid errors like The Avro schema must be provided, designate the location of the schema in your streamConfigs section. For example, if your current section contains the following:

Then add this key: "stream.kafka.decoder.prop.schema"followed by a value that denotes the location of your schema.

Apache Kafka

hashtagCreate schema configuration

hashtagCreate table configuration

hashtagUpload schema and table

hashtagAdd sample data to the Kafka topic

hashtagQuery the table

hashtagKafka ingestion guidelines

hashtagKafka versions in Pinot

hashtagUpgrade from Kafka 0.9 connector to Kafka 2.x connector

hashtagHow to consume from a Kafka version > 2.0.0

hashtagKafka configurations in Pinot

hashtagUse Kafka partition (low) level consumer with SSL

hashtagConsume transactionally-committed messages

hashtagUse Kafka partition (low) level consumer with SASL_SSL

hashtagExtract record headers as Pinot table columns

hashtagTell Pinot where to find an Avro schema

Apache Kafka

hashtagCreate schema configuration

hashtagCreate table configuration

hashtagUpload schema and table

hashtagAdd sample data to the Kafka topic

hashtagQuery the table

hashtagKafka ingestion guidelines

hashtagKafka versions in Pinot

hashtagUpgrade from Kafka 0.9 connector to Kafka 2.x connector

hashtagHow to consume from a Kafka version > 2.0.0

hashtagKafka configurations in Pinot

hashtagUse Kafka partition (low) level consumer with SSL

hashtagConsume transactionally-committed messages

hashtagUse Kafka partition (low) level consumer with SASL_SSL

hashtagExtract record headers as Pinot table columns

hashtagTell Pinot where to find an Avro schema

Create schema configuration

Create table configuration

Upload schema and table

Add sample data to the Kafka topic

Query the table

Kafka ingestion guidelines

Kafka versions in Pinot

Upgrade from Kafka 0.9 connector to Kafka 2.x connector

How to consume from a Kafka version > 2.0.0

Kafka configurations in Pinot

Use Kafka partition (low) level consumer with SSL

Consume transactionally-committed messages

Use Kafka partition (low) level consumer with SASL_SSL

Extract record headers as Pinot table columns

Tell Pinot where to find an Avro schema

Create schema configuration

Create table configuration

Upload schema and table

Add sample data to the Kafka topic

Query the table

Kafka ingestion guidelines

Kafka versions in Pinot

Upgrade from Kafka 0.9 connector to Kafka 2.x connector

How to consume from a Kafka version > 2.0.0

Kafka configurations in Pinot

Use Kafka partition (low) level consumer with SSL

Consume transactionally-committed messages

Use Kafka partition (low) level consumer with SASL_SSL

Extract record headers as Pinot table columns

Tell Pinot where to find an Avro schema