Ingest from Apache Kafka
This guide shows you how to ingest a stream of records from an Apache Kafka topic into a Pinot table.
Learn how to ingest data from Kafka, a stream processing platform. You should have a local cluster up and running, following the instructions in Set up a cluster.
Install and Launch Kafka
Let's start by downloading Kafka to our local machine.
To pull down the latest Docker image, run the following command:
docker pull apache/kafka:4.0.0Download Kafka from kafka.apache.org/quickstart#quickstart_download and then extract it:
tar -xzf kafka_2.13-4.0.0.tgz
cd kafka_2.13-4.0.0Next we'll spin up a Kafka broker. Kafka 4.0 uses KRaft mode by default and does not require ZooKeeper:
docker run --network pinot-demo --name=kafka \
-e KAFKA_NODE_ID=1 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qk \
apache/kafka:4.0.0Note: The --network pinot-demo flag is optional and assumes that you have a Docker network named pinot-demo that you want to connect the Kafka container to.
Kafka 4.0 uses KRaft mode by default. Generate a cluster ID and format the storage directory, then start the broker:
Start Kafka Broker (KRaft mode)
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties
bin/kafka-server-start.sh config/server.propertiesData Source
We're going to generate some JSON messages from the terminal using the following script:
datagen.py
If you run this script (python datagen.py), you'll see the following output:
Ingesting Data into Kafka
Let's now pipe that stream of messages into Kafka, by running the following command:
We can check how many messages have been ingested by running the following command:
Output
And we can print out the messages themselves by running the following command
Output
Schema
A schema defines what fields are present in the table along with their data types in JSON format.
Create a file called /tmp/pinot/schema-stream.json and add the following content to it.
Table Config
A table is a logical abstraction that represents a collection of related data. It is composed of columns and rows (known as documents in Pinot). The table config defines the table's properties in JSON format.
Create a file called /tmp/pinot/table-config-stream.json and add the following content to it.
JSON stream payload formats
org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder still defaults to UTF-8 text JSON when stream.kafka.decoder.prop.jsonFormat is unset, so existing Kafka table configs keep their historical behavior.
To decode a binary JSON stream, add stream.kafka.decoder.prop.jsonFormat to the streamConfigs block and pin the payload encoding:
Supported values are TEXT, POSTGRES_JSONB, SQLITE_JSONB, SMILE, CBOR, and AUTO.
Use AUTO only when a topic can legitimately contain more than one of those encodings. AUTO is opt-in, and its CBOR detection only works when each message includes the CBOR self-describe tag. If the producer always emits one known format, pin that format explicitly instead.
This setting applies only to stream decoding. Batch ingestion with JSONRecordReader still reads text JSON files.
Create schema and table
Create the table and schema by running the appropriate command below:
Querying
Navigate to localhost:9000/#/query and click on the events table to run a query that shows the first 10 rows in this table.
_Querying the events table_
Kafka ingestion guidelines
Kafka connector modules in Pinot
Pinot ships two Kafka connector modules:
pinot-kafka-3.0-- Uses Kafka client library 3.x (currently 3.9.2). This is the default connector included in Pinot distributions. Consumer factory class:org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory.pinot-kafka-4.0-- Uses Kafka client library 4.x (currently 4.1.1). This connector drops the ZooKeeper-based Scala dependency and uses the pure-Java Kafka client, suitable for KRaft-mode Kafka clusters. Consumer factory class:org.apache.pinot.plugin.stream.kafka40.KafkaConsumerFactory.
Migrating from the kafka-2.x connector
If your existing table configs reference the removed kafka-2.x connector, update the stream.kafka.consumer.factory.class.name property:
From:
org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactoryTo (Kafka 3.x):
org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactoryTo (Kafka 4.x):
org.apache.pinot.plugin.stream.kafka40.KafkaConsumerFactory
No other stream config changes are required. The Kafka 3.x connector is compatible with Kafka brokers 2.x and above. The Kafka 4.x connector requires Kafka brokers 4.0 or above.
Kafka configurations in Pinot
Resolve Kafka client secrets with ConfigProvider
The Kafka 3.x and 4.x connectors support Kafka ConfigProvider references in realtime table configs. Use a provider when Kafka client properties, such as SSL passwords, should be loaded when Pinot constructs the Kafka consumer or shared AdminClient instead of being stored directly in the table config.
Declare the provider alias and its parameters in the same streamConfigs or streamConfigMaps object as the Kafka property that references it. This example reads a keystore password from a mounted properties file:
The alias in the reference (file above) must be listed in config.providers; otherwise Pinot processes the value as its own environment-variable or system-property expression. Provider aliases and config.providers.<alias>.* parameters are scoped to their containing stream config map, which is important for multi-stream table configs.
Use Kafka client property names without the stream.kafka.consumer.prop. prefix, such as ssl.keystore.password. That prefix is not a general-purpose namespace for arbitrary Kafka client properties.
Kafka resolves the reference when a client is constructed. Recreating a consumer reads the provider source again, but changing the file does not hot-reload an already running Kafka client.
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.
Use Confluent Schema Registry with JSON encoded messages
If your Kafka messages are JSON-encoded and registered with Confluent Schema Registry, use the KafkaConfluentSchemaRegistryJsonMessageDecoder. This decoder uses the Confluent KafkaJsonSchemaDeserializer to decode messages whose JSON schemas are managed by the registry.
When to use this decoder
Your Kafka producer serializes messages using the Confluent JSON Schema serializer.
Your JSON schemas are registered in Confluent Schema Registry.
You want schema validation and evolution support for JSON messages.
If your messages are Avro-encoded and registered with Schema Registry, use KafkaConfluentSchemaRegistryAvroMessageDecoder instead (shown in the SSL example above). If your messages are plain JSON without a schema registry, use JSONMessageDecoder.
Example table config
The key configuration properties for this decoder are:
stream.kafka.decoder.class.name-- Set toorg.apache.pinot.plugin.inputformat.json.confluent.KafkaConfluentSchemaRegistryJsonMessageDecoder.stream.kafka.decoder.prop.schema.registry.rest.url-- The URL of the Confluent Schema Registry.
Authentication
This decoder supports the same authentication options as the Avro schema registry decoder. You can configure SSL or SASL_SSL authentication for both the Kafka consumer and the Schema Registry client using the stream.kafka.decoder.prop.schema.registry.* properties. See the SSL example and SASL_SSL example above for details.
For Schema Registry basic authentication, add the following properties:
Consume transactionally-committed messages
The Kafka 3.x and 4.x connectors support 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.
Recover from expired Kafka offsets
Kafka can delete records before a lagging Pinot consumer reads them. When the requested offset has expired, Pinot now defaults the Kafka client's raw auto.offset.reset property to earliest. Consumption resumes from the oldest record Kafka still retains, and Pinot's existing stream-data-loss detection reports the unavoidable gap for records Kafka already deleted.
To override this behavior, add the raw Kafka property to streamConfigs:
This raw property accepts Kafka values such as earliest and latest. Do not confuse it with stream.kafka.consumer.prop.auto.offset.reset: the prefixed Pinot property accepts smallest, largest, time periods, or timestamps and chooses the starting position only when a partition has no stored Pinot offset. It does not control Kafka-client recovery from an expired stored offset.
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:
Record key: any type
__key : String
For simplicity of design, we assume that the record key is always a UTF-8 encoded String
Record Headers: Map<String, String>
Each header key is listed as a separate column: __header$HeaderKeyName : String
For simplicity of design, we directly map the string headers from kafka record to pinot table column
Record metadata - offset : long
__metadata$offset : String
Record metadata - partition : int
__metadata$partition : String
Record metadata - recordTimestamp : long
__metadata$recordTimestamp : String
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.
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.
Subset partition ingestion
By default, a Pinot REALTIME table consumes from all partitions of the configured Kafka topic. In some scenarios you may want a table to consume only a subset of the topic's partitions. The stream.kafka.partition.ids setting lets you specify exactly which Kafka partitions a table should consume.
When to use subset partition ingestion
Split-topic ingestion -- Multiple Pinot tables share the same Kafka topic, and each table is responsible for a different set of partitions. This is useful when the same topic contains logically distinct data partitioned by key, and you want separate tables (or indexes) for each partition group.
Multi-table partition assignment -- You want to distribute the partitions of a high-throughput topic across several Pinot tables for workload isolation, independent scaling, or different retention policies.
Selective consumption -- You only need data from specific partitions of a topic (for example, partitions that correspond to a particular region or tenant).
Configuration
Add stream.kafka.partition.ids to the streamConfigMaps entry in your table config. The value can contain individual zero-based partition IDs, inclusive ranges, or a mix of both in a comma-separated string:
For example, "0,2,5" consumes three explicit partitions, "0-3" consumes partitions 0 through 3 inclusive, and "0-3,6,8-9" mixes both forms in one value.
When this setting is present, Pinot will consume only from the resolved set of partitions. When it is absent or blank, Pinot consumes from all partitions of the topic (the default behavior).
If the table also uses segmentPartitionConfig for partition pruning, keep numPartitions aligned with the full Kafka topic partition count, not the size of the consumed subset. Pinot derives realtime segment partition metadata from the full topic partition count even when stream.kafka.partition.ids restricts ingestion to a subset.
Example: splitting a topic across two tables
Suppose you have a Kafka topic called events with two partitions (0 and 1). You can create two Pinot tables, each consuming from one partition:
Table events_part_0:
Table events_part_1:
Validation rules and limitations
Partition IDs must be non-negative integers. Range values are inclusive, and the start of a range must be less than or equal to the end.
Non-integer values (for example,
"abc") and malformed ranges (for example,"5-") will cause a validation error.Duplicate IDs are silently deduplicated after range expansion. For example,
"0,2,0,5"is treated as"0,2,5".The partition IDs are sorted internally for stable ordering, regardless of the order specified in the config.
Pinot validates the resolved partition IDs against the Kafka topic metadata before it starts consuming. If a specified partition ID does not exist in the topic, an error is raised.
The resolved set can contain at most 10,000 unique partition IDs.
If you configure
segmentPartitionConfigon a subset-ingestion realtime table, setnumPartitionsto the full Kafka topic partition count. For example, if a table consumes"0,3"from an 8-partition topic, use8, not2.When using subset partition ingestion with multiple tables consuming from the same topic, ensure that the partition assignments do not overlap if you want each record to be consumed by exactly one table. Pinot does not enforce non-overlapping partition assignments across tables.
Whitespace around partition IDs, commas, and range bounds is trimmed (for example,
" 0 - 3 , 5 "is valid).
Use Protocol Buffers (Protobuf) format
Pinot supports decoding Protocol Buffer messages from Kafka using several decoder options depending on your setup.
ProtoBufMessageDecoder (descriptor file based)
Use ProtoBufMessageDecoder when you have a pre-compiled .desc (descriptor) file for your Protobuf schema. This decoder uses dynamic message parsing and does not require compiled Java classes.
Required stream config properties:
stream.kafka.decoder.prop.descriptorFile
Path or URI to the .desc descriptor file. Supports local file paths, HDFS, and other Pinot-supported file systems.
stream.kafka.decoder.prop.protoClassName
(Optional) Fully qualified Protobuf message name within the descriptor. If omitted, the first message type in the descriptor is used.
stream.kafka.decoder.prop.descriptorFileFallbackEnabled
(Optional) true or false. Overrides the cluster-wide pinot.server.protobuf.descriptor.fallback.enabled setting for this table. The default is true.
Example streamConfigs:
For a remote descriptor URI, Pinot fetches the descriptor afresh whenever this decoder is created, so an in-place descriptor update is picked up at the next consuming-segment transition. If that fetch fails, Pinot can reuse the last descriptor content that fetched and resolved successfully for the same URI on that server JVM. This fallback is enabled by default and logs a warning when used. It does not help a new server process that has never loaded the descriptor, and it does not apply to local descriptor files.
Pinot does not fall back when the remote fetch succeeds but the descriptor is empty, corrupt, or lacks the configured message type; decoder initialization fails so a bad schema deployment remains visible. If you prefer to fail ingestion on any remote fetch failure, set stream.kafka.decoder.prop.descriptorFileFallbackEnabled to false in the table's streamConfigs, or disable the cluster setting. The table setting takes precedence. This fallback applies only to ProtoBufMessageDecoder, not the compiled-JAR or batch Protobuf readers.
ProtoBufCodeGenMessageDecoder (compiled JAR based)
Use ProtoBufCodeGenMessageDecoder when you have a compiled JAR containing your generated Protobuf Java classes. This decoder uses runtime code generation for improved decoding performance.
Required stream config properties:
stream.kafka.decoder.prop.jarFile
Path or URI to the JAR file containing compiled Protobuf classes.
stream.kafka.decoder.prop.protoClassName
Fully qualified Java class name of the Protobuf message (required).
Example streamConfigs:
KafkaConfluentSchemaRegistryProtoBufMessageDecoder (Confluent Schema Registry)
Use KafkaConfluentSchemaRegistryProtoBufMessageDecoder when your Protobuf schemas are managed by Confluent Schema Registry. This decoder automatically resolves schemas from the registry at runtime.
Required stream config properties:
stream.kafka.decoder.prop.schema.registry.rest.url
URL of the Confluent Schema Registry.
Optional properties:
stream.kafka.decoder.prop.cached.schema.map.capacity
Maximum number of cached schemas. Default: 1000.
stream.kafka.decoder.prop.schema.registry.*
SSL and authentication options for connecting to Schema Registry (same pattern as the Avro Confluent decoder).
Example streamConfigs:
Use BSON format
Pinot supports decoding BSON messages from Kafka using BSONMessageDecoder. Use this when each Kafka message contains a single binary-encoded BSON document, such as records produced by a MongoDB change-data-capture pipeline.
Example streamConfigs:
BSONMessageDecoder uses the same extractor as batch BSON ingestion, so ObjectId values decode to hex strings, Date and BsonTimestamp values decode to java.sql.Timestamp, Decimal128 values decode to BigDecimal with NaN and Infinity surfaced as null, binary values decode to byte[], embedded documents decode to Map<String, Object>, and arrays decode to Object[].
Use Apache Arrow format
Pinot supports decoding Apache Arrow IPC streaming format messages from Kafka using ArrowMessageDecoder. This is useful when upstream systems produce data serialized in Arrow format.
Optional stream config properties:
stream.kafka.decoder.prop.arrow.allocator.limit
Maximum memory (in bytes) for the Arrow allocator. Default: 268435456 (256 MB).
stream.kafka.decoder.prop.extractRawTimeValues
Keep Arrow Date, Time, and Timestamp values as raw integers instead of Pinot's default converted values. Default: false.
Example streamConfigs:
When stream.kafka.decoder.prop.extractRawTimeValues is false (the default), Pinot converts Arrow Date, Time, and Timestamp values during extraction. Set it to true to keep raw integers instead: Date stays as days since epoch, while Time and Timestamp stay in the schema's declared Arrow unit.
Each Arrow Kafka message can yield zero, one, or many Pinot rows. Empty batches are ignored, single-row batches ingest as one Pinot row, and multi-row batches fan out into multiple Pinot rows.
Consuming a Subset of Kafka Partitions
By default, a Pinot realtime table consumes all partitions of a Kafka topic. You can restrict ingestion to a specific subset of partitions using the stream.kafka.partition.ids property. This is useful when:
Splitting a single Kafka topic across multiple Pinot tables for independent scaling
Multi-tenant scenarios where different tables own different partition ranges
Configuration
Add stream.kafka.partition.ids to your streamConfigs with a comma-separated string of partition IDs, inclusive ranges, or both:
For example, "0,2,5" selects individual partitions, "0-3" selects an inclusive range, and "0-3,6,8-9" mixes both forms in one value.
If the table also uses segmentPartitionConfig, keep numPartitions set to the full Kafka topic partition count rather than the size of the consumed subset. Pinot computes realtime segment partition metadata from the full topic partition count even when ingestion is limited to specific Kafka partitions.
Notes
Partition IDs must be non-negative integers. Ranges are inclusive and must have
start <= end.Pinot validates the resolved partition IDs against the Kafka topic metadata before it starts consuming.
Duplicate IDs in the list are automatically deduplicated.
The resolved set can contain at most 10,000 unique partition IDs.
The total partition count reported to the broker reflects the full Kafka topic size, ensuring correct query routing across tables sharing the same topic.
If you configure
segmentPartitionConfig, setnumPartitionsto the full Kafka topic partition count. For example, if a table consumes"1,4"from a 6-partition topic, use6, not2.When splitting a topic between two tables, configure one with even-numbered IDs and another with odd-numbered IDs (for example,
"0,2"and"1,3"for a 4-partition topic).
Last updated
Was this helpful?