Streaming de datos en tiempo real con Python: Kafka y Redis Streams
El streaming de datos en tiempo real procesa eventos a medida que se generan, sin esperar a almacenar lotes. Python ofrece excelentes clientes para Apache Kafka (kafka-python, confluent-kafka) y Redis Streams (redis-py). Esta guía cubre los patrones esenciales: productor/consumidor, grupos de consumidores y procesamiento con asyncio.
Apache Kafka con Python
Instalación
pip install kafka-python confluent-kafka
# Para Avro/Schema Registry:
pip install confluent-kafka[avro]
Productor básico
from kafka import KafkaProducer
import json
import time
from datetime import datetime
# Configurar productor
productor = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode('utf-8'),
key_serializer=lambda k: k.encode('utf-8') if k else None,
acks='all', # esperar confirmación de todos los réplicas
retries=3,
max_in_flight_requests_per_connection=1,
)
def enviar_evento(topic, clave, datos):
"""Envía un evento al topic de Kafka."""
futuro = productor.send(
topic,
key=clave,
value=datos,
)
try:
metadata = futuro.get(timeout=10)
print(f"Enviado → {topic} partición={metadata.partition} offset={metadata.offset}")
return True
except Exception as e:
print(f"Error enviando: {e}")
return False
# Simular flujo de eventos de sensor IoT
while True:
evento = {
"dispositivo_id": "sensor-001",
"temperatura": 22.5 + (time.time() % 5),
"humedad": 65.0,
"timestamp": datetime.utcnow().isoformat(),
"unidad": "°C",
}
enviar_evento("sensores-iot", "sensor-001", evento)
time.sleep(1)
productor.flush()
productor.close()
Consumidor básico
from kafka import KafkaConsumer
import json
consumidor = KafkaConsumer(
'sensores-iot',
bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest', # 'latest' para solo mensajes nuevos
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='procesador-sensores', # grupo de consumidores
max_poll_records=100,
)
print("Escuchando mensajes...")
for mensaje in consumidor:
evento = mensaje.value
print(f"[{mensaje.topic}:{mensaje.partition}:{mensaje.offset}] "
f"Sensor={evento['dispositivo_id']} "
f"Temp={evento['temperatura']:.1f}°C")
# Procesar el evento
if evento['temperatura'] > 25.0:
print(f" ⚠️ ALERTA: temperatura elevada en {evento['dispositivo_id']}")
Grupo de consumidores (balanceo de carga)
from kafka import KafkaConsumer
import json
import threading
def consumidor_worker(worker_id, topic, group_id):
"""Cada worker en un grupo consume particiones distintas."""
consumidor = 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} iniciado")
for mensaje in consumidor:
print(f"Worker {worker_id}: procesando {mensaje.key} offset={mensaje.offset}")
# ... procesar mensaje
# Lanzar 3 consumidores en paralelo (cada uno procesa una partición)
workers = []
for i in range(3):
t = threading.Thread(
target=consumidor_worker,
args=(i, 'pedidos', 'grupo-procesador-pedidos')
)
t.daemon = True
t.start()
workers.append(t)
for w in workers:
w.join()
Commit manual de offsets
from kafka import KafkaConsumer, TopicPartition
import json
consumidor = KafkaConsumer(
'transacciones',
bootstrap_servers=['localhost:9092'],
group_id='auditoria',
enable_auto_commit=False, # commit manual
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
)
for mensaje in consumidor:
try:
# Procesar
procesar_transaccion(mensaje.value)
# Solo hacer commit si el procesamiento fue exitoso
consumidor.commit({
TopicPartition(mensaje.topic, mensaje.partition):
mensaje.offset + 1 # próximo offset a consumir
})
except Exception as e:
print(f"Error procesando: {e} — no se hizo commit")
def procesar_transaccion(datos):
print(f"Procesando: {datos}")
# lógica de negocio
Productor asíncrono con confluent-kafka
from confluent_kafka import Producer
import json
config = {
'bootstrap.servers': 'localhost:9092',
'acks': 'all',
'retries': 5,
'batch.size': 16384,
'linger.ms': 5, # agrupar mensajes en ventana de 5ms
'compression.type': 'snappy',
}
productor = Producer(config)
def callback_entrega(err, msg):
if err:
print(f"Error: {err}")
else:
print(f"Entregado: {msg.topic()} [{msg.partition()}] @ {msg.offset()}")
# Producir en lote
for i in range(1000):
productor.produce(
topic='eventos-masivos',
key=f"key-{i}",
value=json.dumps({"id": i, "valor": i * 3.14}),
callback=callback_entrega,
)
if i % 100 == 0:
productor.poll(0) # servir callbacks pendientes
productor.flush() # esperar a que todos los mensajes se entreguen
Redis Streams
Redis Streams (desde Redis 5.0) es una alternativa ligera a Kafka para volúmenes menores.
Instalación
pip install redis
Productor con Redis Streams
import redis
import json
import time
from datetime import datetime
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def publicar_evento(stream, datos, max_len=10000):
"""
Publica en un Redis Stream.
max_len: tamaño máximo del stream (mensajes más antiguos se eliminan)
"""
id_mensaje = r.xadd(
stream,
datos,
maxlen=max_len,
approximate=True # ~ para eficiencia
)
return id_mensaje
# Publicar eventos
for i in range(10):
id_msg = publicar_evento('stream:pedidos', {
'pedido_id': f'PED-{i:05d}',
'cliente': f'cliente_{i}',
'total': str(100 * (i + 1)),
'timestamp': datetime.utcnow().isoformat(),
})
print(f"Publicado: {id_msg}")
time.sleep(0.1)
Consumidor simple
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def consumir_stream(stream, ultimo_id='$', bloquear_ms=1000):
"""
ultimo_id:
- '$' = solo mensajes nuevos
- '0' = desde el principio
- id específico = desde ese ID
"""
print(f"Consumiendo de '{stream}' desde ID={ultimo_id}")
while True:
mensajes = r.xread(
{stream: ultimo_id},
count=10,
block=bloquear_ms
)
if not mensajes:
continue
for nombre_stream, lista_msgs in mensajes:
for id_msg, datos in lista_msgs:
print(f"ID={id_msg}: {datos}")
ultimo_id = id_msg # avanzar cursor
consumir_stream('stream:pedidos', ultimo_id='0')
Grupos de consumidores con Redis Streams
import redis
import threading
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
# Crear grupo de consumidores (solo una vez)
try:
r.xgroup_create('stream:pedidos', 'grupo-procesadores', id='0', mkstream=True)
print("Grupo creado")
except redis.ResponseError as e:
if 'BUSYGROUP' not in str(e):
raise
def consumidor_grupo(nombre_consumidor):
"""
Los mensajes se reparten entre consumidores del mismo grupo.
Cada mensaje se entrega a un solo consumidor.
"""
while True:
mensajes = r.xreadgroup(
groupname='grupo-procesadores',
consumername=nombre_consumidor,
streams={'stream:pedidos': '>'}, # '>' = nuevos mensajes no entregados
count=5,
block=2000
)
if not mensajes:
continue
for nombre_stream, lista in mensajes:
for id_msg, datos in lista:
try:
print(f"{nombre_consumidor} procesando {id_msg}: {datos['pedido_id']}")
# ... procesar el pedido
# Confirmar que se procesó (ACK)
r.xack('stream:pedidos', 'grupo-procesadores', id_msg)
except Exception as e:
print(f"{nombre_consumidor} error: {e} — sin ACK (se reintentará)")
# Lanzar 3 consumidores en paralelo
for i in range(3):
t = threading.Thread(target=consumidor_grupo, args=(f"worker-{i}",))
t.daemon = True
t.start()
Recuperar mensajes no confirmados (pending)
import redis
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
def recuperar_pendientes(stream, grupo, min_idle_ms=60000):
"""
Recupera mensajes que llevan más de min_idle_ms sin confirmar
(posiblemente de un consumidor que falló).
"""
pendientes = r.xpending_range(
stream, grupo,
min='-', max='+',
count=100,
idle=min_idle_ms
)
for msg in pendientes:
id_msg = msg['message_id']
consumidor = msg['consumer']
tiempo_ms = msg['time_since_delivered']
intentos = msg['times_delivered']
print(f"Pendiente: {id_msg} de {consumidor} "
f"({tiempo_ms}ms, {intentos} intentos)")
if intentos > 3:
# Mover a DLQ (Dead Letter Queue)
r.xadd('stream:pedidos:dlq',
{'id_original': id_msg, 'motivo': 'demasiados intentos'})
r.xack(stream, grupo, id_msg)
else:
# Reclamar para este consumidor
r.xclaim(stream, grupo, 'worker-recuperador', min_idle_ms, [id_msg])
recuperar_pendientes('stream:pedidos', 'grupo-procesadores')
asyncio para streaming concurrente
import asyncio
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json
async def productor_async(topic, n_mensajes=100):
productor = AIOKafkaProducer(
bootstrap_servers='localhost:9092',
value_serializer=lambda v: json.dumps(v).encode(),
)
await productor.start()
try:
for i in range(n_mensajes):
await productor.send(topic, {"id": i, "valor": i ** 2})
await asyncio.sleep(0.01)
print(f"Enviados {n_mensajes} mensajes")
finally:
await productor.stop()
async def consumidor_async(topic):
consumidor = AIOKafkaConsumer(
topic,
bootstrap_servers='localhost:9092',
group_id='async-group',
value_deserializer=lambda m: json.loads(m.decode()),
)
await consumidor.start()
try:
async for msg in consumidor:
print(f"Recibido: {msg.value}")
finally:
await consumidor.stop()
async def main():
# Ejecutar productor y consumidor concurrentemente
await asyncio.gather(
productor_async('test-topic'),
consumidor_async('test-topic'),
)
asyncio.run(main())
Recurso adicional
Para convertir archivos de datos entre CSV, JSON, Parquet y otros formatos sin necesidad de programar, usa KaijuConverter — gratis y sin registro.
Conversiones relacionadas
Conversiones de vídeo que encajan con esta guía: