push to kafka with mysql data using python and python kafka client sdk

 https://medium.com/@sangeethaprabhagaran/pushing-mysql-data-into-kafka-6ab8ab5bf755

kafka does not support http request

hence any kafka push pull need to go through developed SDK or u make an SDK based on kafka documentaion on your own

push to kafka using python sdk from mysql db example:

Prerequisites:-

• Install MySQL on your laptop, create an employee table, and import data into it. You can refer to my other post
• Install Kafka on your laptop, and create an employee topic. You can refer to my other post
• Install the MySQL connector by executing the below command in your command prompt —

pip install mysql-connector-python

• Once the above installation successful, then install the kafka python by executing the below command in your command prompt —

pip install kafka-python

• As a next step you can verify these installs in your python shell as below

Execution:

• We will try to read the MySQL table data in to cursor convert it in to Json format and send the records to the Kafka topic over a loop. We will also import dumps from the Json library, so that we can serialize our value (records) into Json.

• Initialize the connection to your MySQL database — development. You can use your username and password here.

• Set the cursor to fetch all the records from the employee table —

• Create an instance of the KafkaProducer client with the basic configurations. Please ensure to start your zookeeper and Kafka server before executing this code. We are trying to connect to our local Kafka server and setting a JSON serializer for value.In Kafka data is stored as key, value pairs. However, having key is optional. So we are skipping the key here. Also, there are many other serializers available. Please refer to Kafka-python to get more options.

• Next step is to send the MySQL records in a JSON format to the Kafka topic — employee, we have created already

• Once you execute the below code you will see the below output, which are the records from the MySQL employee table with serialization applied.

• To verify the records you can run a Kafka consumer as below

bin\windows\kafka-console-consumer.bat — bootstrap-server localhost:9092 — topic employee — from-beginning

I had run the push multiple times and hence you see duplicate messages here. Also, we have used “from-beginning” option to read all the messages. In case we want to look at only the latest messages, then we can skip this option -

That’s it. We have pushed the MySQL data into the Kafka topic. You can access this complete code in my github

kafka replication of brokers, consumer group, consumer, topics and partitions

topics and paritions:

https://codingharbour.com/apache-kafka/the-introduction-to-kafka-topics-and-partitions/

Partition has several purposes in Kafka.

From Kafka broker’s point of view, partitions allow a single topic to be distributed over multiple servers. That way it is possible to store more data in a topic than what a single server could hold. If you imagine you needed to store 10TB of data in a topic and you have 3 brokers, one option would be to create a topic with one partition and store all 10TB on one broker. Another option would be to create a topic with 3 partitions and spread 10 TB of data over all the brokers. 

replication of brokers:

https://stackoverflow.com/questions/44787552/in-kafka-is-each-message-replicated-across-all-partitions-of-a-topic#:~:text=Each%20message%20goes%20into%20a,is%20replicated%20across%20those%20brokers.

Replication does not occur across partitions. Each message goes into a single partition of the topic, no matter how many partitions the topic has.

If you have set the replication-factor for topic to a number larger than 1 (assuming you have multiple brokers running in the cluster), then each partition of the topic is replicated across those brokers.

consumer groups:

onsumer group

A consumer group is a group of consumers (I guess you didn’t see this coming?) that share the same group id. When a topic is consumed by consumers in the same group, every record will be delivered to only one consumer. As the official documentation states: “If all the consumer instances have the same consumer group, then the records will effectively be load-balanced over the consumer instances.”

This way you can ensure parallel processing of records from a topic and be sure that your consumers won’t be stepping on each other toes.

How does Kafka achieve this?

Each topic consists of one or more partitions. When a new consumer is started it will join a consumer group (this happens under the hood) and Kafka will then ensure that each partition is consumed by only one consumer from that group.

So, if you have a topic with two partitions and only one consumer in a group, that consumer would consume records from both partitions.

A single consumer in a consumer group

https://stackoverflow.com/questions/69866354/how-can-you-maintain-ordering-at-the-consumer-level-in-kafka

If only 1partion, 3 consumer, 2 consumer will waiat, if 3 parition 3 consumer, they will each consume 1, then swap

Kafka getting "org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1195725856 larger than 104857600)"

 https://stackoverflow.com/questions/49370959/getting-org-apache-kafka-common-network-invalidreceiveexception-invalid-receiv

1195725856 is GET[space] encoded as a big-endian, four-byte integer (see here for more information on how that works). This indicates that HTTP traffic is being sent to Kafka port 9092, but Kafka doesn't accept HTTP traffic, it only accepts its own protocol (which takes the first four bytes as the receive size, hence the error).

Since the error is received on startup, it is likely benign and may indicate a scanning service or similar on your network scanning ports with protocols that Kafka doesn't understand.

In order to find the cause, you can find where the HTTP traffic is coming from using tcpdump: