Real-Time Data Streaming with Python: Kafka and Redis Streams
Real-time data streaming processes events as they occur, without waiting to batch them. Python provides excellent clients for Apache Kafka (kafka-python, confluent-kafka) and Redis Streams (redis-py). This guide covers essential patterns: producer/consumer, consumer groups, manual commits, and async processing.
Apache Kafka with Python
Installation
pip install kafka-python confluent-kafka
# For Avro/Schema Registry:
pip install confluent-kafka[avro]
Basic Producer
from kafka import KafkaProducer
import json
import time
from datetime import datetime
producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
key_serializer=lambda k: k.encode('utf-8') if k else None,
acks='all',
retries=3,
max_in_flight_requests_per_connection=1,
)
def send_event(topic, key, data):
future = producer.send(topic, key=key, value=data)
try:
meta = future.get(timeout=10)
print(f"Sent → {topic} partition={meta.partition} offset={meta.offset}")
return True
except Exception as e:
print(f"Send error: {e}")
return False
# Simulate IoT sensor event stream
while True:
event = {
"device_id": "sensor-001",
"temperature": 22.5 + (time.time() % 5),
"humidity": 65.0,
"timestamp": datetime.utcnow().isoformat(),
"unit": "C",
}
send_event("iot-sensors", "sensor-001", event)
time.sleep(1)
producer.flush()
producer.close()
Basic Consumer
from kafka import KafkaConsumer
import json
consumer = KafkaConsumer(
'iot-sensors',
bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest', # 'latest' for only new messages
enable_auto_commit=True,
auto_commit_interval_ms=1000,
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
key_deserializer=lambda k: k.decode('utf-8') if k else None,
group_id='sensor-processor',
max_poll_records=100,
)
print("Listening for messages...")
for message in consumer:
event = message.value
print(f"[{message.topic}:{message.partition}:{message.offset}] "
f"Device={event['device_id']} Temp={event['temperature']:.1f}C")
if event['temperature'] > 25.0:
print(f" ⚠️ ALERT: high temperature on {event['device_id']}")
Consumer Group (Load Balancing)
from kafka import KafkaConsumer
import json
import threading
def consumer_worker(worker_id, topic, group_id):
"""Each worker in a group consumes different partitions."""
consumer = KafkaConsumer(
topic,
bootstrap_servers=['localhost:9092'],
group_id=group_id,
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
auto_offset_reset='latest',
)
print(f"Worker {worker_id} started")
for message in consumer:
print(f"Worker {worker_id}: processing {message.key} offset={message.offset}")
# Launch 3 parallel consumers (each processes separate partitions)
workers = []
for i in range(3):
t = threading.Thread(
target=consumer_worker,
args=(i, 'orders', 'order-processor-group')
)
t.daemon = True
t.start()
workers.append(t)
for w in workers:
w.join()
Manual Offset Commits
from kafka import KafkaConsumer, TopicPartition
import json
consumer = KafkaConsumer(
'transactions',
bootstrap_servers=['localhost:9092'],
group_id='audit-group',
enable_auto_commit=False,
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
)
for message in consumer:
try:
process_transaction(message.value)
# Only commit if processing succeeded
consumer.commit({
TopicPartition(message.topic, message.partition):
message.offset + 1
})
except Exception as e:
print(f"Processing error: {e} — skipping commit")
def process_transaction(data):
print(f"Processing: {data}")
High-Performance Async Producer (confluent-kafka)
from confluent_kafka import Producer
import json
config = {
'bootstrap.servers': 'localhost:9092',
'acks': 'all',
'retries': 5,
'batch.size': 16384,
'linger.ms': 5,
'compression.type': 'snappy',
}
producer = Producer(config)
def delivery_callback(err, msg):
if err:
print(f"Error: {err}")
else:
print(f"Delivered: {msg.topic()} [{msg.partition()}] @ {msg.offset()}")
for i in range(1000):
producer.produce(
topic='bulk-events',
key=f"key-{i}",
value=json.dumps({"id": i, "value": i * 3.14}),
callback=delivery_callback,
)
if i % 100 == 0:
producer.poll(0) # serve pending callbacks
producer.flush()
Redis Streams
Redis Streams (since Redis 5.0) is a lightweight Kafka alternative for smaller volumes.
Installation
pip install redis
Redis Streams Producer
import redis
import time
from datetime import datetime
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def publish_event(stream, data, max_len=10000):
msg_id = r.xadd(stream, data, maxlen=max_len, approximate=True)
return msg_id
for i in range(10):
msg_id = publish_event('stream:orders', {
'order_id': f'ORD-{i:05d}',
'customer': f'customer_{i}',
'total': str(100 * (i + 1)),
'timestamp': datetime.utcnow().isoformat(),
})
print(f"Published: {msg_id}")
time.sleep(0.1)
Simple Consumer
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def consume_stream(stream, last_id='$', block_ms=1000):
"""
last_id:
- '$' = only new messages
- '0' = from the beginning
- specific ID = from that ID forward
"""
while True:
messages = r.xread({stream: last_id}, count=10, block=block_ms)
if not messages:
continue
for stream_name, msg_list in messages:
for msg_id, data in msg_list:
print(f"ID={msg_id}: {data}")
last_id = msg_id
consume_stream('stream:orders', last_id='0')
Consumer Groups with Redis Streams
import redis
import threading
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
try:
r.xgroup_create('stream:orders', 'processor-group', id='0', mkstream=True)
print("Consumer group created")
except redis.ResponseError as e:
if 'BUSYGROUP' not in str(e):
raise
def group_consumer(consumer_name):
"""Messages are distributed — each message goes to exactly one consumer."""
while True:
messages = r.xreadgroup(
groupname='processor-group',
consumername=consumer_name,
streams={'stream:orders': '>'}, # '>' = new undelivered messages
count=5,
block=2000
)
if not messages:
continue
for stream_name, msg_list in messages:
for msg_id, data in msg_list:
try:
print(f"{consumer_name} processing {msg_id}: {data['order_id']}")
# ... process order
# Acknowledge successful processing
r.xack('stream:orders', 'processor-group', msg_id)
except Exception as e:
print(f"{consumer_name} error: {e} — no ACK (will retry)")
for i in range(3):
t = threading.Thread(target=group_consumer, args=(f"worker-{i}",))
t.daemon = True
t.start()
Recover Unacknowledged Messages
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def recover_pending(stream, group, min_idle_ms=60000):
"""Recover messages idle for more than min_idle_ms (from failed consumers)."""
pending = r.xpending_range(
stream, group, min='-', max='+',
count=100, idle=min_idle_ms
)
for msg in pending:
msg_id = msg['message_id']
consumer = msg['consumer']
idle_ms = msg['time_since_delivered']
attempts = msg['times_delivered']
print(f"Pending: {msg_id} from {consumer} ({idle_ms}ms, {attempts} attempts)")
if attempts > 3:
# Move to Dead Letter Queue
r.xadd('stream:orders:dlq',
{'original_id': msg_id, 'reason': 'too many attempts'})
r.xack(stream, group, msg_id)
else:
# Claim for recovery worker
r.xclaim(stream, group, 'recovery-worker', min_idle_ms, [msg_id])
recover_pending('stream:orders', 'processor-group')
Async Streaming with asyncio
pip install aiokafka
import asyncio
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json
async def async_producer(topic, count=100):
producer = AIOKafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.dumps(v).encode(),
)
await producer.start()
try:
for i in range(count):
await producer.send(topic, {"id": i, "value": i ** 2})
await asyncio.sleep(0.01)
print(f"Sent {count} messages")
finally:
await producer.stop()
async def async_consumer(topic):
consumer = AIOKafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
group_id='async-group',
value_deserializer=lambda m: json.loads(m.decode()),
)
await consumer.start()
try:
async for msg in consumer:
print(f"Received: {msg.value}")
finally:
await consumer.stop()
async def main():
await asyncio.gather(
async_producer('test-topic'),
async_consumer('test-topic'),
)
asyncio.run(main())
Additional Resource
For converting data files between CSV, JSON, Parquet and other formats without any coding, use KaijuConverter — free and no registration required.
Related conversions
Common video conversions that pair well with this guide: