Streaming real: procesamiento continuo de eventos

1from kafka import KafkaConsumer, KafkaProducer
2import json
3
4consumer = KafkaConsumer(
5    "pedidos-raw",
6    bootstrap_servers=["localhost:9092"],
7    group_id="transformador-pedidos",
8    auto_offset_reset="earliest",
9    enable_auto_commit=False,
10    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
11)
12
13producer = KafkaProducer(
14    bootstrap_servers=["localhost:9092"],
15    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
16)
17
18def transformar_pedido(raw: dict) -> dict:
19    """Enriquece y normaliza un pedido raw."""
20    return {
21        "pedido_id": raw["pedido_id"],
22        "cliente_id": raw["cliente_id"],
23        "total_con_iva": round(raw["total"] * 1.21, 2),
24        "moneda": "EUR",
25        "categoria": clasificar_total(raw["total"]),
26        "procesado_at": time.time(),
27    }
28
29def clasificar_total(total: float) -> str:
30    if total > 500: return "premium"
31    if total > 100: return "standard"
32    return "micro"
33
34print("Stream processor: pedidos-raw → pedidos-enriched")
35for msg in consumer:
36    enriched = transformar_pedido(msg.value)
37    producer.send("pedidos-enriched", value=enriched)
38    consumer.commit()
39    print(f"  {enriched['pedido_id']}: {enriched['total_con_iva']}EUR ({enriched['categoria']})")

Transformación en vuelo: leer de un topic, transformar, escribir en otro

1# Filtrado: solo pedidos VIP (> 500 EUR) van al topic destino
2for msg in consumer:
3    pedido = msg.value
4    if pedido["total"] > 500:
5        producer.send("pedidos-vip", value=pedido)
6        print(f"  VIP: {pedido['pedido_id']} -> {pedido['total']}EUR")
7    else:
8        print(f"  Skip: {pedido['pedido_id']} ({pedido['total']}EUR)")
9    consumer.commit()

Filtrado: solo los eventos que cumplen la condición pasan al topic destino

1import time
2from collections import defaultdict
3
4class VentanaTemporal:
5    """Agregación por ventana de tiempo sliding window."""
6    
7    def __init__(self, window_seconds: int = 60):
8        self.window_seconds = window_seconds
9        self.eventos = []  # (timestamp, valor)
10    
11    def agregar(self, valor: float, timestamp: float = None):
12        ts = timestamp or time.time()
13        self.eventos.append((ts, valor))
14        self._limpiar_expirados()
15    
16    def _limpiar_expirados(self):
17        ahora = time.time()
18        self.eventos = [(ts, v) for ts, v in self.eventos
19                       if ahora - ts <= self.window_seconds]
20    
21    def total(self) -> float:
22        self._limpiar_expirados()
23        return sum(v for _, v in self.eventos)
24    
25    def count(self) -> int:
26        self._limpiar_expirados()
27        return len(self.eventos)
28    
29    def promedio(self) -> float:
30        c = self.count()
31        return self.total() / c if c > 0 else 0
32
33# Uso en un stream processor
34ventas_por_tienda = defaultdict(lambda: VentanaTemporal(window_seconds=300))
35
36for msg in consumer:
37    venta = msg.value
38    tienda = venta["tienda_id"]
39    ventas_por_tienda[tienda].agregar(venta["precio"])
40    
41    # Emitir métrica cada mensaje (o cada N mensajes)
42    print(f"  [{tienda}] Últimos 5min: {ventas_por_tienda[tienda].count()} ventas, "
43          f"total: {ventas_por_tienda[tienda].total():.2f}EUR")
44    consumer.commit()

Ventana temporal: agrega datos de los últimos N segundos/minutos

1# Simulamos una "base de datos" de clientes (en producción sería Redis o PostgreSQL)
2clientes_db = {
3    "CLI-1": {"nombre": "María García", "segmento": "premium", "antiguedad_meses": 24},
4    "CLI-2": {"nombre": "Carlos López", "segmento": "standard", "antiguedad_meses": 6},
5    "CLI-3": {"nombre": "Ana Martínez", "segmento": "nuevo", "antiguedad_meses": 1},
6}
7
8def enriquecer_pedido(pedido: dict) -> dict:
9    """Enriquece un pedido con datos del cliente."""
10    cliente_info = clientes_db.get(pedido["cliente_id"], {})
11    return {
12        **pedido,
13        "cliente_nombre": cliente_info.get("nombre", "Desconocido"),
14        "cliente_segmento": cliente_info.get("segmento", "unknown"),
15        "cliente_antiguedad": cliente_info.get("antiguedad_meses", 0),
16        "es_cliente_leal": cliente_info.get("antiguedad_meses", 0) > 12,
17    }
18
19# En el stream processor:
20for msg in consumer:
21    pedido_enriched = enriquecer_pedido(msg.value)
22    producer.send("pedidos-enriched", value=pedido_enriched)
23    print(f"  {pedido_enriched['pedido_id']}: {pedido_enriched['cliente_nombre']} "
24          f"({pedido_enriched['cliente_segmento']})")

Enriquecimiento: completar el evento con datos de una fuente externa

1from collections import defaultdict
2import time
3
4class DetectorAnomalias:
5    """Detecta anomalías basadas en frecuencia en ventana de tiempo."""
6    
7    def __init__(self, max_eventos: int = 5, ventana_seg: int = 120):
8        self.max_eventos = max_eventos
9        self.ventana_seg = ventana_seg
10        self.historial = defaultdict(list)  # key -> [timestamps]
11    
12    def registrar(self, key: str) -> bool:
13        """Registra un evento y retorna True si es anómalo."""
14        ahora = time.time()
15        # Limpiar eventos fuera de la ventana
16        self.historial[key] = [
17            ts for ts in self.historial[key]
18            if ahora - ts <= self.ventana_seg
19        ]
20        # Agregar nuevo evento
21        self.historial[key].append(ahora)
22        # ¿Supera el umbral?
23        return len(self.historial[key]) > self.max_eventos
24
25# Uso en el stream processor
26detector = DetectorAnomalias(max_eventos=5, ventana_seg=120)
27
28for msg in consumer:
29    pedido = msg.value
30    cliente = pedido["cliente_id"]
31    
32    if detector.registrar(cliente):
33        # ALERTA: posible fraude
34        alerta = {
35            "tipo": "frecuencia_alta",
36            "cliente_id": cliente,
37            "pedidos_en_ventana": len(detector.historial[cliente]),
38            "timestamp": time.time(),
39        }
40        producer.send("alertas-fraude", value=alerta)
41        print(f"  🚨 ALERTA FRAUDE: {cliente} ({len(detector.historial[cliente])} pedidos en 2min)")
42    
43    consumer.commit()

Detección de anomalías: alertar cuando la frecuencia supera un umbral