Stream ingestion example

Search…
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
Batch import example
HDFS as Deep Storage
Last modified 30d ago
Copy link
Contents
Data Stream
Creating a Schema
Creating a table config
Uploading your schema and table config
Loading sample data into stream
Ingesting streaming data