[Python] Properly Setup Kafka With Python for performance and scalability

I'm new to Apache Kafka and want to set it up with Python to track vehicle locations.

My plan is to use WebSockets to send continuous coordinate data from the frontend to the backend. For WebSocket implementation, I'm using Django Channels. While this setup allows me to receive real-time data from the frontend, inserting each incoming record directly into PostgreSQL is not ideal for performance and scalability.

Initially, I considered using Redis for this, but after researching best practices for vehicle tracking, I found that Kafka is a more suitable choice.

Everything is working fine so far. However, I don't want to insert data into the database continuously for every incoming message. To address this, I implemented batch processing in my Kafka consumer. Despite this, the consumer still receives messages immediately after they are produced.

I need guidance on how to properly configure Kafka for my use case so that messages are processed in batches before being inserted into PostgreSQL. Below is the sample code for my Kafka producer and consumer.

consumers.py

KAFKA_TOPIC = 'vehicle-location-updates'
KAFKA_BROKER = 'localhost:9092' # Adjust to your Kafka broker

# Create Kafka consumer
consumer = KafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=[KAFKA_BROKER],
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

def consume_vehicle_locations():
for message in consumer:
# Get the data from the Kafka message
data = message.value
vehicle_id = data.get('vehicle_id')
lat = data.get('latitude')
lon = data.get('longitude')
print(f"Location updated for {vehicle_id} at {lat, lon}")


KAFKA_TOPIC = 'vehicle-location-updates'
KAFKA_BROKER = 'localhost:9092' # Adjust to your Kafka broker

# Create Kafka consumer
consumer = KafkaConsumer(
KAFKA_TOPIC,
bootstrap_servers=[KAFKA_BROKER],
value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

def consume_vehicle_locations():
for message in consumer:
# Get the data from the Kafka message
data = message.value
vehicle_id = data.get('vehicle_id')
lat = data.get('latitude')
lon = data.get('longitude')
print(f"Location updated for {vehicle_id} at {lat, lon}")


producers.py
from kafka import KafkaProducer
import json

KAFKA_TOPIC = "vehicle-location-updates"
KAFKA_BROKER = "localhost:9092" # Adjust to your Kafka broker


# Custom partitioner to ensure messages for the same vehicle_id go to the same partition
def vehicle_partition(key_bytes, all_partitions, available_partitions):
vehicle_id = key_bytes.decode('utf-8')
# Use the vehicle_id hash to determine the partition (this is just an example)
partition = hash(vehicle_id) % len(available_partitions)
return partition

# Kafka Producer Configuration
producer = KafkaProducer(
bootstrap_servers=[KAFKA_BROKER],
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
key_serializer=lambda v: str(v).encode('utf-8'), # Serialize vehicle_id as key
batch_size=1048576, # 1 MB batch size (increase to accumulate more messages)
linger_ms=100000, # Wait for 100 ms before sending the batch
partitioner=vehicle_partition # Custom partitioner for vehicles
)

# Send location messages for multiple vehicles
def send_vehicle_location(vehicle_id, lat, lon):
message = {
'vehicle_id': vehicle_id,
'latitude': lat,
'longitude': lon
}

# Send the message with the vehicle_id as the key (to ensure same vehicle goes to same partition)
producer.send(
KAFKA_TOPIC,
value=message,
key=str(vehicle_id) # Send vehicle_id as key to ensure partitioning
)

producer.flush()

# Example usage
for i in range(10):
send_vehicle_location(vehicle_id=1, lat=40.7128, lon=-74.0060)
send_vehicle_location(vehicle_id=2, lat=34.0522, lon=-118.2437)

Continue reading...