Basic Kafka Stream in Python
Tags: python, kafka, programming
Published: 2023-03-11 19:25:02

# Introduction

Kafka is an open-source distributed streaming platform developed by the Apache Software Foundation. It is designed to handle large volumes of real-time data streams and is used for **building real-time data pipelines and streaming applications**.

For instance, if you would like to have a real-time event log fetch from the source (Twitter, etc) and insert it into the target (CSV, SQLite, etc), you can leverage the Kafka library in Python. So, you need to know what are producer and consumer in Kafka.

### Producer and Consumer

![Producer & Consumer in Kafka](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/haeeggx890re937suv4g.png)

1. **Producer** is a program or application that sends data or messages to a Kafka topic. 
2. **Consumer** is a program or application that reads data or messages from a Kafka topic.


# How to setup Kafka in Python?
Now, let's start with requirements to configure Kafka in Python.

## Requirements

### Kafka server 
You can start with local installation or AWS and GCP for advanced method). In this tutorial, I am launching Kafka server using [Docker Compose](https://docs.docker.com/compose/).

Assume Docker is installed and running or run `docker --version` to confirm

Create a docker compose file (**docker-compose.yaml**) and copy below snippet.

```yaml
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 22181:2181
  
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
```

Then, we could build the Docker
`docker-compose up --build --force-recreate -d`

To confirm the containers are up, we need to check the status.

**Docker Desktop**
![Docker container status](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/bczuj7edexvqlaz41y98.png)

**Command**
```
╰─➤  docker-compose ps
      Name                   Command            State                      Ports
---------------------------------------------------------------------------------------------------
kafka_kafka_1       /etc/confluent/docker/run   Up      0.0.0.0:29092->29092/tcp, 9092/tcp
kafka_zookeeper_1   /etc/confluent/docker/run   Up      0.0.0.0:22181->2181/tcp, 2888/tcp, 3888/tcp
```

Okay, now it shows the container is successfully built and running. Next step, how to create a topic and send the message?

### Kafka producer (Python) 

#### Create a topic

```python
from confluent_kafka.admin import AdminClient, NewTopic

admin_client = AdminClient({"bootstrap.servers": "localhost:29092"})

topic_name = "my_first_topic"
num_partitions = 3

new_topic = NewTopic(topic_name, num_partitions)

try:
    admin_client.create_topics([new_topic])
    print(f"{topic_name} is created!!")
except:
    print(f"{topic_name} is not created!!")
```

I am using `confluent_kafka` to integrate with Kafka server.

Also, we can define the number of partitions for the topic. Important to know that number of partitions cannot be changed after the topic has been created.

If you have a topic with 10 partitions, you can have up to 10 consumers processing messages from that topic in parallel, each handling messages from its assigned partition.

```
╰─➤  python kafka_topic.py
my_first_topic is created!!
```

Okay, the topic has been created!

> Note: If you first forgot to create a topic, no worries. It should create when you post the message to the new topic.


#### Post a message

```python
import json
import uuid
from confluent_kafka import Producer


def post_message(producer, data):
    try:
        producer.produce("my_first_topic", json.dumps(data).encode("utf-8"))
        producer.flush()
        print(f"message posted!! --> {data['comment']}")
    except:
        print("failed to post message")


data = {
    "user_session_id": str(uuid.uuid4()),
    "user_name": "John Doe",
    "comment": "Malaysia Boleh!",
}

producer = Producer({"bootstrap.servers": "localhost:29092"})
post_message(producer, data)
```

When you run the Python script, it could post the message

```
╰─➤  python producer.py
message posted!! --> Malaysia Boleh!
```


### Kafka consumer (Python) 
Okay, let's move to the consumer to check whether the data is really posted.


```python
from confluent_kafka import Consumer

consumer = Consumer(
    {
        "bootstrap.servers": "localhost:29092",
        "group.id": "python-consumer",
        "auto.offset.reset": "earliest",
    }
)

consumer.subscribe(["my_first_topic"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print("Error: {}".format(msg.error()))
        continue
    data = msg.value().decode("utf-8")
    topic = msg.topic()
    ts = msg.timestamp()
    print(data, topic, ts)
```

And, here is the output when you run the script

```
╰─➤  python consumer.py
{"user_session_id": "5321202b-694a-4d82-9d2b-97174303595e", "user_name": "John Doe", "comment": "Malaysia Boleh!"} my_first_topic (1, 1678561082329)
```
Here is the demo when you post a message, the consumer can print it out within a second.

[![Demo](https://share.gifyoutube.com/lR9DVl.gif)](https://#)

Enjoy ~