Deserialization in Kafka - Mini Project: Build & Apply

Deserialization with Kafka Consumer

📖 Scenario: You are working with Kafka to read messages from a topic. These messages are sent as JSON strings. To use them in your program, you need to convert these JSON strings back into Python dictionaries. This process is called deserialization.

🎯 Goal: Build a simple Kafka consumer that reads JSON messages from a topic and deserializes them into Python dictionaries.

📋 What You'll Learn

Create a Kafka consumer that connects to a local Kafka server

Configure the consumer to read from the topic named test-topic

Deserialize the JSON string messages into Python dictionaries

Print the deserialized dictionary for each message

💡 Why This Matters

🌍 Real World

Kafka is widely used to stream data between systems. Deserialization is essential to convert raw message bytes into usable data structures.

💼 Career

Understanding Kafka deserialization is important for roles in data engineering, backend development, and real-time data processing.

1

Set up Kafka consumer

Import KafkaConsumer from kafka and create a consumer called consumer that connects to localhost:9092 and subscribes to the topic 'test-topic'.

Kafka

# Import KafkaConsumer and create consumer
# Your code here

Need a hint?

Use KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092']) to create the consumer.

2

Add JSON deserialization setup

Import the json module and add a variable called decode_utf8 that decodes bytes to UTF-8 strings. This will help convert message values from bytes to strings before deserialization.

Kafka

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'])

# Create decode_utf8 function to decode bytes to string
# Your code here

Need a hint?

Define a function decode_utf8 that takes data and returns data.decode('utf-8').

3

Deserialize messages in the consumer loop

Use a for loop with variable message to iterate over consumer. Inside the loop, decode the message value using decode_utf8 and then deserialize it using json.loads into a variable called data.

Kafka

from kafka import KafkaConsumer
import json

def decode_utf8(data):
    return data.decode('utf-8')

consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'])

# Loop over consumer and deserialize messages
# Your code here

Need a hint?

Use for message in consumer: and inside decode with decode_utf8(message.value) then deserialize with json.loads.

4

Print the deserialized data

Inside the for loop, add a print statement to display the data dictionary.

Kafka

from kafka import KafkaConsumer
import json

def decode_utf8(data):
    return data.decode('utf-8')

consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'])

for message in consumer:
    json_str = decode_utf8(message.value)
    data = json.loads(json_str)
    # Print the deserialized data
    # Your code here

Need a hint?

Use print(data) inside the loop to show the dictionary.