Apache Pinot Docs
Search…
Stream ingestion example
The Docker instructions on this page are still WIP
So far, we setup our cluster, ran some queries on the demo tables and explored the admin endpoints. We also uploaded some sample batch data for transcript table.
Now, it's time to ingest from a sample stream into Pinot. The rest of the instructions assume you're using Pinot in Docker.

Data Stream

First, we need to setup a stream. Pinot has out-of-the-box realtime ingestion support for Kafka. Other streams can be plugged in, more details in Pluggable Streams.
Let's setup a demo Kafka cluster locally, and create a sample topic transcript-topic
Docker
Using launcher scripts
Start Kafka
1
docker run \
2
--network pinot-demo_default --name=kafka \
3
-e KAFKA_ZOOKEEPER_CONNECT=manual-zookeeper:2181/kafka \
4
-e KAFKA_BROKER_ID=0 \
5
-e KAFKA_ADVERTISED_HOST_NAME=kafka \
6
-d wurstmeister/kafka:latest
Copied!
Create a Kafka Topic
1
docker exec \
2
-t kafka \
3
/opt/kafka/bin/kafka-topics.sh \
4
--zookeeper manual-zookeeper:2181/kafka \
5
--partitions=1 --replication-factor=1 \
6
--create --topic transcript-topic
Copied!
Start Kafka
Start Kafka cluster on port 9876 using the same Zookeeper from the quick-start examples
1
bin/pinot-admin.sh StartKafka -zkAddress=localhost:2123/kafka -port 9876
Copied!
Create a Kafka topic
Download the latest Kafka. Create a topic
1
bin/kafka-topics.sh --create --bootstrap-server localhost:9876 --replication-factor 1 --partitions 1 --topic transcript-topic
Copied!

Creating a Schema

If you followed the Batch upload sample data, you have already pushed a schema for your sample table. If not, head over to Creating a schema on that page, to learn how to create a schema for your sample data.

Creating a table config

If you followed Batch upload sample data, you learnt how to push an offline table and schema. Similar to the offline table config, we will create a realtime table config for the sample. Here's the realtime table config for the transcript table. For a more detailed overview about table, checkout Table.
/tmp/pinot-quick-start/transcript-table-realtime.json
1
{
2
"tableName": "transcript",
3
"tableType": "REALTIME",
4
"segmentsConfig": {
5
"timeColumnName": "timestampInEpoch",
6
"timeType": "MILLISECONDS",
7
"schemaName": "transcript",
8
"replicasPerPartition": "1"
9
},
10
"tenants": {},
11
"tableIndexConfig": {
12
"loadMode": "MMAP",
13
"streamConfigs": {
14
"streamType": "kafka",
15
"stream.kafka.consumer.type": "lowlevel",
16
"stream.kafka.topic.name": "transcript-topic",
17
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
18
"stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
19
"stream.kafka.broker.list": "kafka:9092",
20
"realtime.segment.flush.threshold.rows": "0",
21
"realtime.segment.flush.threshold.time": "24h",
22
"realtime.segment.flush.threshold.segment.size": "50M",
23
"stream.kafka.consumer.prop.auto.offset.reset": "smallest"
24
}
25
},
26
"metadata": {
27
"customConfigs": {}
28
}
29
}
Copied!

Uploading your schema and table config

Now that we have our table and schema, let's upload them to the cluster. As soon as the realtime table is created, it will begin ingesting from the Kafka topic.
Docker
Launcher Script
1
docker run \
2
--network=pinot-demo_default \
3
-v /tmp/pinot-quick-start:/tmp/pinot-quick-start \
4
--name pinot-streaming-table-creation \
5
apachepinot/pinot:latest AddTable \
6
-schemaFile /tmp/pinot-quick-start/transcript-schema.json \
7
-tableConfigFile /tmp/pinot-quick-start/transcript-table-realtime.json \
8
-controllerHost manual-pinot-controller \
9
-controllerPort 9000 \
10
-exec
Copied!
1
bin/pinot-admin.sh AddTable \
2
-schemaFile /tmp/pinot-quick-start/transcript-schema.json \
3
-tableConfigFile /tmp/pinot-quick-start/transcript-table-realtime.json \
4
-exec
Copied!

Loading sample data into stream

Here's a JSON file for transcript table data:
/tmp/pinot-quick-start/rawData/transcript.json
1
{"studentID":205,"firstName":"Natalie","lastName":"Jones","gender":"Female","subject":"Maths","score":3.8,"timestampInEpoch":1571900400000}
2
{"studentID":205,"firstName":"Natalie","lastName":"Jones","gender":"Female","subject":"History","score":3.5,"timestampInEpoch":1571900400000}
3
{"studentID":207,"firstName":"Bob","lastName":"Lewis","gender":"Male","subject":"Maths","score":3.2,"timestampInEpoch":1571900400000}
4
{"studentID":207,"firstName":"Bob","lastName":"Lewis","gender":"Male","subject":"Chemistry","score":3.6,"timestampInEpoch":1572418800000}
5
{"studentID":209,"firstName":"Jane","lastName":"Doe","gender":"Female","subject":"Geography","score":3.8,"timestampInEpoch":1572505200000}
6
{"studentID":209,"firstName":"Jane","lastName":"Doe","gender":"Female","subject":"English","score":3.5,"timestampInEpoch":1572505200000}
7
{"studentID":209,"firstName":"Jane","lastName":"Doe","gender":"Female","subject":"Maths","score":3.2,"timestampInEpoch":1572678000000}
8
{"studentID":209,"firstName":"Jane","lastName":"Doe","gender":"Female","subject":"Physics","score":3.6,"timestampInEpoch":1572678000000}
9
{"studentID":211,"firstName":"John","lastName":"Doe","gender":"Male","subject":"Maths","score":3.8,"timestampInEpoch":1572678000000}
10
{"studentID":211,"firstName":"John","lastName":"Doe","gender":"Male","subject":"English","score":3.5,"timestampInEpoch":1572678000000}
11
{"studentID":211,"firstName":"John","lastName":"Doe","gender":"Male","subject":"History","score":3.2,"timestampInEpoch":1572854400000}
12
{"studentID":212,"firstName":"Nick","lastName":"Young","gender":"Male","subject":"History","score":3.6,"timestampInEpoch":1572854400000}
Copied!
Push sample JSON into Kafka topic, using the Kafka script from the Kafka download
1
bin/kafka-console-producer.sh \
2
--broker-list localhost:9876 \
3
--topic transcript-topic < /tmp/pinot-quick-start/rawData/transcript.json
Copied!

Ingesting streaming data

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 Query Console to checkout the realtime data
Last modified 30d ago