How to consume message form kafka topic

Photo by Cytonn Photography on Unsplash

To consume messages from a Kafka topic using Python, you can again use the kafka-python library. Below is a simple example that demonstrates how to set up a Kafka consumer:

Step 1: Install kafka-python

If you haven’t already installed the kafka-python library, you can do so using pip:

pip install kafka-python

Step 2: Consumer Code

Here’s a basic Python script to consume messages from a Kafka topic:

from kafka import KafkaConsumer
import json

# Define the Kafka topic and broker address
KAFKA_BROKER = 'localhost:9092'  # Adjust this if your broker is on a different host
TOPIC_NAME = 'my-topic'
# Create a Kafka consumer
consumer = KafkaConsumer(
    TOPIC_NAME,
    bootstrap_servers=[KAFKA_BROKER],
    value_deserializer=lambda x: json.loads(x.decode('utf-8')),  # Deserialize messages from JSON format
    auto_offset_reset='earliest',  # Start reading at the earliest message
    enable_auto_commit=True,
    group_id='my-group'  # Specify a consumer group
)
# Consume messages from the Kafka topic
try:
    for message in consumer:
        print(f"Received message: {message.value}")
finally:
    # Close the consumer
    consumer.close()

Explanation:

• KafkaConsumer: This class is used to create a Kafka consumer instance. You provide it with the Kafka broker’s address, topic name, and deserialization method for the messages.
• bootstrap_servers: This parameter specifies the Kafka broker address.
• value_deserializer: This function is used to convert the byte messages back into Python objects. Here, it deserializes JSON-formatted messages.
• auto_offset_reset='earliest': This setting tells the consumer to start reading messages from the beginning of the topic if no committed offsets exist.
• enable_auto_commit=True: This setting enables automatic offset committing, meaning the consumer will keep track of the last read offset.
• group_id: This parameter specifies the consumer group for your consumer. Multiple consumers in the same group will share the load of reading messages.

Step 3: Running the Script

Run the script, and it will start consuming messages from the specified Kafka topic. Each message received will be printed to the console.

Make sure your Kafka broker is running, and the topic exists before you run the consumer.