AWS Managed Streaming for Kafka

 


AWS Managed Streaming for Kafka

(Govern how your clients interact with Apache Kafka using API Gateway)

       

Watch my You tube video on this:


Follow the steps below to create the data pipeline:

STEP1

Networking Part

VPC , Public and Private Subnets, Internet Gateway, NAT Gateway , S3 end point , Route Tables , Security Groups



STEP2 

Launch MSK Cluster in the same VPC as created in STEP1 , unauthorized access allowed , plaintext encryption
Make sure the security group remain as it is.

STEP3

Launch Linux EC2
Choose the Same VPC , create in Public Subnet
In the list Auto-assign Public IP, choose Enable.

STEP4

Once the client for Amazon MSK has been created, the security group rules must be configured to allow the connection between the cluster and the 
client ec2 machine that we have just created.

For that , Add the security group id of ec2 to MSK cluster security group all traffic

Repeat these steps to add an inbound rule in the security group that corresponds to your client computer to allow it to receive traffic 
from the security group from the VPC. Now your client computer can communicate bidirectionally with the MSK Cluster.

Once this is done, the newly created and configured client can be accessed.

STEP5

Login to EC2 machine and do the JAVA installation as below:

sudo dnf update -y
sudo dnf install java-17-amazon-corretto-devel
java -version


STEP6

Install Kafka as well using below commands:


wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
tar -xvf kafka_2.12-2.8.1.tgz
cd kafka_2.12-2.8.1


Create Topic

bin/kafka-topics.sh --create --topic datapunditdemo1 --bootstrap-server b-1.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092,b-2.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092 --replication-factor 1 --partitions 1

bin/kafka-topics.sh --create --topic datapunditdemo --bootstrap-server b-1.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092,b-2.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092 --replication-factor 1 --partitions 1

bin/kafka-console-producer.sh --topic datapunditdemo --bootstrap-server {Place the MSK bootstrap server URLs here}
bin/kafka-console-producer.sh --topic datapunditdemo --bootstrap-server b-1.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092,b-2.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092

bin/kafka-console-consumer.sh --topic datapunditdemo --bootstrap-server {Place the MSK bootstrap server URLs here}
bin/kafka-console-consumer.sh --topic datapunditdemo --bootstrap-server b-1.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092,b-2.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092
 
 
STEP7 confluent kafka
Install confluent kafka within kafka_2.12-2.8.1)
wget  http://packages.confluent.io/archive/5.1/confluent-5.1.2-2.11.zip
unzip confluent-5.1.2-2.11.zip

export CONFLUENT_HOME=/home/ec2-user/kafka_2.12-2.8.1/confluent-5.1.2
export PATH=$PATH:$CONFLUENT_HOME/bin

STEP8
Change the bootstrap.servers in  confluent-5.1.2/etc/kafka-rest/kafka-rest.properties

Replace bootstrap.servers=PLAINTEXT://localhost
by the below
bootstrap.servers=PLAINTEXT://b-1.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092,PLAINTEXT://b-2.testcluster1.8m8ze0.c9.kafka.us-east-1.amazonaws.com:9092


STEP9
Start the Kafka Rest
/home/ec2-user/kafka_2.12-2.8.1/confluent-5.1.2/bin/kafka-rest-start /home/ec2-user/kafka_2.12-2.8.1/confluent-5.1.2/etc/kafka-rest/kafka-rest.properties 


Content-Type application/vnd.kafka.json.v2+json

http://44.204.68.5:8082/topics/datapunditdemo

Sample Data to post:

{"records":[{"value":{"name": "User1"}}]}

STEP10

Configure API Gateway to secure API call
https://aws.amazon.com/blogs/big-data/govern-how-your-clients-interact-with-apache-kafka-using-api-gateway/


http://44.204.68.5:8082/:8082/{proxy}

Use the following secure to test the API using postman
https://muceuuyk91.execute-api.us-east-1.amazonaws.com/dev/datapunditdemo



For more Abinitio, AWS and data engineering videos please subscribe , view , like and share my YouTube channel 

Click DataPundit

Comments

Post a Comment

Popular posts from this blog

Abinitio Interview Question # 1 - Write Multiple Files in Abinitio

Image
 Question :  I get records of various region in huge volume ,  we want to write the records in region specific files.   How to achieve that? View the explanation in the below video: Answer: -WMF WMF: filename::get_filename(in)= begin let string("") l_filename=string_substring(in.data.region); filename::string_concat($AI_SERIAL,"/",l_filename.dat); end; out::reformat(in)= begin out::in.data; end; OUTPUT DML-: DML record string(int) filename; string(integer(4)) file_contents; end; RMF: filename::get_filename(in)= begin filename::string_concat($AI_SERIAL,"/",in.filename); end; out::reformat(read,filename,in)= begin out.file_contents::read; out.filename::filename; end; Please view my youtube channel  DataPundit
Read more
Image
  Metaprogramming in Abinitio 1. How to create output DML dynamically based in INPUT DML 2. How to create intermediate transform( XFR) dynamically based on INPUT DML      Please See my you tube video for explanation : PART 1 PART 2 For Example, let us say, we have an Input DML as below-: cat input.dml record string(",") name; decimal(",") age; decimal(",") salary; string(",") aadharid; date("yyyy-mm-dd")(",") joiningdate; string(",") mobileno=NULL; string(",") dlno=NULL; string(",") passportnumber=NULL; end; (1.1) prepare a new DML based on input DML i.e. add fields, remove fields, modify type Graph based parameters: ===================================================================== input_dml=$DIR/input.dml  [interpretation = constant] input_dml_info =$[read_type(input_dml)]   [interpretation = pdl] /* Define functions in AB_DML_DEFS */ AB_DML_DEFS= $[ out:: dynamic_dml_gen(input_dml_info)= beg
Read more

Next In Sequence in ABinitio | How next_in_sequence() works in MFS

Image
https://youtu.be/JdDDpjZcv-4 next_in_sequence() working in Abinitio: Definition -      within a single process(component) , this gives successive number.   next_in_sequence()>1 i.e. all records whose occurance is 2 and more in the file.      for example dataset below:    employeeName ram shyam ravi hanuman shekhar seeta geeta shayama vidushi vishakha If we process by condition next_in_sequence()>1 then the output will be [first record will be removed] shyam ravi hanuman shekhar seeta geeta shayama vidushi vishakha 2. next_in_sequence() with reformat  - rely -  FieldA 1a 2a 3a 4a 5a 6a 7a Example 1 (serial processing)--> Reformat   count=2      select:   next_in_sequence()<900      Transofom1:   out:reformat(in)=   begin   out.*::in.*;   out.Seq-No::next_in_sequence();   end;      Transform2:   out:reformat(in)=   begin   out.*::in.*;   out.Seq-No::next_in_sequence();   end;      Output_indexes()=   begin   out::if(next_in_sequence()%2==0) [vector 0,1] else [vector 0,1];   e
Read more