Apache Pinot Docs
Search…
latest
For Users
For Developers
For Operators
Configuration Reference
Amazon MSK (Kafka)
How to Connect Pinot with Amazon Managed Streaming for Apache Kafka (Amazon MSK)
Prerequisite
Create an Amazon MSK Cluster
Note:
- 1.For demo simplicity, this MSK cluster reuses same VPC created by EKS cluster in the previous step. Otherwise a VPC Peering is required to ensure two VPCs could talk to each other.
- 2.Under Encryption section, choose
Both TLS encrypted and plaintext traffic allowed
Below is a sample screenshot to create an Amazon MSK cluster.
1
Copied!
After click on Create button, you can take a coffee break and come back.
Amazon MSK Clusters View
Once the cluster is created, you can view it and click
View client information to see the Zookeeper and Kafka Broker list.
MSK Cluster View
Sample Client Information
Connect to MSK
Config SecurityGroup
Until now, the MSK cluster is still not accessible, you can follow this Wiki to create an EC2 instance to connect to it for topic creation, run console producer and consumer.
In order to connect MSK to EKS, we need to allow the traffic could go through each other.
This is configured through Amazon VPC Page.
- 1.Record the Amazon MSK
SecurityGroupfrom the Cluster page, in the above demo, it'ssg-01e7ab1320a77f1a9. - 2.Open Amazon VPC Page, click on
SecurityGroupson left bar. Find the EKS Security group:eksctl-${PINOT_EKS_CLUSTER}-cluster/ClusterSharedNodeSecurityGroup.
Amazon EKS ClusterSharedNodeSecurityGroup
Please ensure you are picking ClusterShardNodeSecurityGroup
- 1.In SecurityGroup, click on MSK SecurityGroup (
sg-01e7ab1320a77f1a9), then Click onEdit Rules, then add aboveClusterSharedNodeSecurityGroup(sg-0402b59d7e440f8d1) to it.
Add SecurityGroup to Amazon MSK
- 1.Click EKS Security Group
ClusterSharedNodeSecurityGroup(sg-0402b59d7e440f8d1), add In bound Rule for MSK Security Group (sg-01e7ab1320a77f1a9).
Add SecurityGroup to Amazon EKS
Now, EKS cluster should be able to talk to Amazon MSK.
Create Kafka topic
To run below commands, please ensure you set two environment variable with
ZOOKEEPER_CONNECT_STRING and BROKER_LIST_STRING (Use plaintext) from Amazon MSK client information, and replace the Variables accordingly.E.g.
1
ZOOKEEPER_CONNECT_STRING="z-3.pinot-quickstart-msk-d.ky727f.c3.kafka.us-west-2.amazonaws.com:2181,z-1.pinot-quickstart-msk-d.ky727f.c3.kafka.us-west-2.amazonaws.com:2181,z-2.pinot-quickstart-msk-d.ky727f.c3.kafka.us-west-2.amazonaws.com:2181"
2
BROKER_LIST_STRING="b-1.pinot-quickstart-msk-d.ky727f.c3.kafka.us-west-2.amazonaws.com:9092,b-2.pinot-quickstart-msk-d.ky727f.c3.kafka.us-west-2.amazonaws.com:9092"
Copied!
You can log into one EKS node or container and run below command to create a topic.
E.g. Enter into Pinot controller container:
1
kubectl exec -it pod/pinot-controller-0 -n pinot-quickstart bash
Copied!
Then install
wget then download Kafka binary.1
apt-get update
2
apt-get install wget -y
3
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
4
tar -xzf kafka_2.12-2.2.1.tgz
5
cd kafka_2.12-2.2.1
Copied!
Create a Kafka topic:
1
bin/kafka-topics.sh \
2
--zookeeper ${ZOOKEEPER_CONNECT_STRING} \
3
--create \
4
--topic pullRequestMergedEventsAwsMskDemo \
5
--replication-factor 1 \
6
--partitions 1
Copied!
Topic creation succeeds with below message:
1
Created topic "pullRequestMergedEventsAwsMskDemo".
Copied!
Write sample data into Kafka
Once topic is created, we can start a simple application to produce to it.
You can download below yaml file, then replace:
${ZOOKEEPER_CONNECT_STRING}-> MSK Zookeeper String${BROKER_LIST_STRING}-> MSK Plaintext Broker String in the deployment${GITHUB_PERSONAL_ACCESS_TOKEN}-> A personal Github Personal Access Token generated from here, please grant all read permissions to it. Here is the source code to generate Github Events.
github-events-aws-msk-demo.yaml
7KB
Code
github-events-aws-msk-demo.yaml
And apply the YAML file by.
1
kubectl apply -f github-events-aws-msk-demo.yaml
Copied!
Once the pod is up, you can verify by running a console consumer to read from it.
Try to run from the Pinot Controller container entered in above step.
1
bin/kafka-console-consumer.sh \
2
--bootstrap-server ${BROKER_LIST_STRING} \
3
--topic pullRequestMergedEventsAwsMskDemo
Copied!
Create a Pinot table
This step is relatively easy.
Since we already put table creation request into the ConfigMap, we can just enter into
pinot-github-events-data-into-msk-kafka pod to execute the command.- Check if the pod is running:
1
kubectl get pod -n pinot-quickstart |grep pinot-github-events-data-into-msk-kafka
Copied!
Sample output:
1
pinot-github-events-data-into-msk-kafka-68948fb4cd-rrzlf 1/1 Running 0 14m
Copied!
- Enter into the pod
1
podname=`kubectl get pod -n pinot-quickstart |grep pinot-github-events-data-into-msk-kafka|awk '{print $1}'`
2
kubectl exec -it ${podname} -n pinot-quickstart bash
Copied!
- Create Table
1
bin/pinot-admin.sh AddTable \
2
-controllerHost pinot-controller \
3
-tableConfigFile /var/pinot/examples/pullRequestMergedEventsAwsMskDemo_realtime_table_config.json \
4
-schemaFile /var/pinot/examples/pullRequestMergedEventsAwsMskDemo_schema.json \
5
-exec
Copied!
Sample output:
1
Executing command: AddTable -tableConfigFile /var/pinot/examples/pullRequestMergedEventsAwsMskDemo_realtime_table_config.json -schemaFile /var/pinot/examples/pullRequestMergedEventsAwsMskDemo_schema.json -controllerHost pinot-controller -controllerPort 9000 -exec
2
Sending request: http://pinot-controller:9000/schemas to controller: pinot-controller-0.pinot-controller-headless.pinot-quickstart.svc.cluster.local, version: Unknown
3
{"status":"Table pullRequestMergedEventsAwsMskDemo_REALTIME succesfully added"}
Copied!
- Then you can open Pinot Query Console to browse the data
Last modified 8mo ago