Detección de Anomalías en Tiempo Real con Apache Kafka y Python

Detectar que algo está saliendo mal — un servidor con latencias inusuales, un sensor industrial fuera de rango, transacciones financieras sospechosas — vale mucho más cuando se detecta en el momento en que ocurre y no horas después revisando logs. Este artículo muestra cómo construir un sistema de detección de anomalías en tiempo real combinando Apache Kafka como infraestructura de mensajería y un modelo de Isolation Forest de scikit-learn que evalúa cada evento al vuelo.

El código completo está disponible en GitHub.

Arquitectura del Sistema

El sistema sigue un pipeline de tres etapas:

1. Entrenamiento: entrenar un modelo de Isolation Forest con datos normales y persistirlo en disco.
2. Producción de datos: un productor genera transacciones en streaming y las publica en un topic de Kafka.
3. Detección: uno o más consumidores leen el topic, aplican el modelo y publican las anomalías detectadas en un topic separado.

El desacoplamiento entre productor y consumidor es una ventaja clave: puedes añadir nuevos consumidores (un segundo modelo, un escritor a S3, un notificador de Slack) sin tocar el productor.

Conceptos de Apache Kafka

Antes de escribir código, los bloques fundamentales de Kafka:

• Broker: servidor Kafka que almacena y entrega mensajes. En producción hay varios formando un clúster.
• Topic: canal de mensajes nombrado. Los productores escriben en topics; los consumidores leen de ellos.
• Partición: un topic se divide en particiones que permiten paralelismo. Cada mensaje va a una partición.
• Productor: cliente que publica mensajes en un topic.
• Consumidor: cliente que se suscribe a un topic y lee sus mensajes.
• Grupo de consumidores: múltiples consumidores cooperan para procesar un topic en paralelo; Kafka balancea particiones entre ellos.
• Offset: posición de un mensaje dentro de una partición. Los consumidores guardan el último offset leído para reanudar sin perder mensajes.

Configurando Kafka con Docker Compose

La forma más rápida de tener Kafka corriendo localmente es con Docker Compose. Usamos la imagen de Confluent Platform que ya incluye todo lo necesario:

# docker-compose.yml
version: "3.8"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

Levanta el clúster:

docker-compose up -d

Instala las dependencias Python:

pip install confluent-kafka scikit-learn numpy joblib

Entrenando el Detector de Anomalías

Isolation Forest es un algoritmo no supervisado ideal para detección de anomalías: construye árboles de aislamiento y asigna puntuaciones de anomalía (más negativo = más anómalo). No necesita ejemplos etiquetados de anomalías para entrenarse.

Este es el script exacto del repositorio, que entrena sobre datos sintéticos 2D con distribución gaussiana:

# model/train.py
import numpy as np
from joblib import dump
from sklearn.ensemble import IsolationForest

rng = np.random.RandomState(42)

# Generar datos de entrenamiento: dos clusters gaussianos (simulan transacciones normales)
X = 0.3 * rng.randn(500, 2)
X_train = np.r_[X + 2, X - 2]   # 1000 muestras totales
X_train = np.round(X_train, 3)

# Entrenar Isolation Forest
clf = IsolationForest(
    n_estimators=50,
    max_samples=500,
    random_state=rng,
    contamination=0.01  # esperamos ~1% de anomalías
)
clf.fit(X_train)

# Persistir el modelo
dump(clf, './isolation_forest.joblib')
print("Modelo guardado en isolation_forest.joblib")

Ejecuta el entrenamiento primero:

python model/train.py

Productor: Generando el Stream de Transacciones

El productor genera transacciones en un loop infinito. La mayoría son datos normales (dos clusters gaussianos); con una probabilidad configurable genera observaciones anómalas (valores uniformes en un rango amplio) para probar el detector.

# streaming/producer.py
import json
import random
import time
from datetime import datetime
import numpy as np
from confluent_kafka import Producer

TRANSACTIONS_TOPIC = "transactions"
OUTLIERS_PROB = 0.05   # 5% de probabilidad de anomalía
DELAY = 0.1            # segundos entre mensajes

producer = Producer({"bootstrap.servers": "localhost:9092"})

_id = 0
print(f"Productor iniciado. Enviando al topic '{TRANSACTIONS_TOPIC}'...")

while True:
    # Generar observación normal o anómala
    if random.random() <= OUTLIERS_PROB:
        # Anomalía: valores uniformes fuera del rango normal
        X_test = np.random.uniform(low=-4, high=4, size=(1, 2))
    else:
        # Normal: cluster gaussiano centrado en +2 o -2
        X = 0.3 * np.random.randn(1, 2)
        X_test = X + np.random.choice(a=[2, -2], size=1, p=[0.5, 0.5])

    X_test = np.round(X_test, 3).tolist()
    current_time = datetime.utcnow().isoformat()
    record = {"id": _id, "data": X_test, "current_time": current_time}

    producer.produce(
        topic=TRANSACTIONS_TOPIC,
        value=json.dumps(record).encode("utf-8")
    )
    producer.flush()

    _id += 1
    time.sleep(DELAY)

Consumidor: Detectando Anomalías al Vuelo

El consumidor se suscribe al topic de transacciones, aplica el modelo a cada mensaje y publica las anomalías detectadas en un topic separado. Para mayor throughput, lanza un proceso por partición del topic:

# streaming/anomalies_detector.py
import json
import os
import logging
from multiprocessing import Process

import numpy as np
from joblib import load
from confluent_kafka import Consumer, Producer

TRANSACTIONS_TOPIC = "transactions"
ANOMALIES_TOPIC = "anomalies"
NUM_PARTITIONS = 3

model_path = os.path.abspath("model/isolation_forest.joblib")


def detect():
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector-group",
        "auto.offset.reset": "latest",
    })
    producer = Producer({"bootstrap.servers": "localhost:9092"})
    clf = load(model_path)

    consumer.subscribe([TRANSACTIONS_TOPIC])

    while True:
        message = consumer.poll(timeout=0.05)  # 50 ms

        if message is None:
            continue
        if message.error():
            logging.error("Consumer error: %s", message.error())
            continue

        record = json.loads(message.value().decode("utf-8"))
        data = record["data"]
        prediction = clf.predict(data)

        if prediction[0] == -1:
            score = clf.score_samples(data)
            record["score"] = np.round(score, 3).tolist()
            producer.produce(
                topic=ANOMALIES_TOPIC,
                value=json.dumps(record).encode("utf-8")
            )
            producer.flush()

    consumer.close()


# Lanzar un proceso por partición para procesar en paralelo
if __name__ == "__main__":
    for _ in range(NUM_PARTITIONS):
        p = Process(target=detect)
        p.start()

El sistema lanza NUM_PARTITIONS procesos en paralelo, uno por partición del topic. Cada proceso tiene su propio consumidor y comparte el mismo grupo de consumidores (anomaly-detector-group), por lo que Kafka balancea automáticamente qué partición lee cada proceso.

Por Qué Importa la Detección en Tiempo Real

El procesamiento por lotes (batch) tiene un costo inherente: el tiempo entre que ocurre un evento y cuando lo detectas. Si tu pipeline de ML corre cada hora, una anomalía puede estar causando daño durante 59 minutos antes de que recibas una alerta. En sistemas críticos — manufactura, finanzas, infraestructura de TI — ese retardo no es aceptable.

El enfoque de streaming invierte la ecuación: en lugar de traer datos al modelo periódicamente, el modelo espera a que lleguen los datos y los evalúa de inmediato. Es la imagen inversa del patrón request-response de servir un modelo de ML con FastAPI y Docker: allí, un cliente pide una predicción; aquí, las predicciones ocurren a medida que llegan los eventos.

Ejecutando el Sistema Completo

# Terminal 1: levantar Kafka
docker-compose up -d

# Terminal 2: entrenar el modelo
python model/train.py

# Terminal 3: iniciar el detector (consumidor)
python streaming/anomalies_detector.py

# Terminal 4: iniciar el productor
python streaming/producer.py

En la consola del detector verás los IDs de transacciones anómalas y sus scores. Los mensajes marcados como anomalías quedan en el topic anomalies para ser consumidos por cualquier otro sistema (Slack, base de datos, dashboard).

Extensiones para Producción

Con este sistema básico funcionando, los próximos pasos naturales son:

• Alertas externas: añade un consumidor del topic anomalies que envíe notificaciones a Slack, PagerDuty o email.
• Reentrenamiento periódico: mantén una ventana deslizante de datos recientes y reentrena el modelo para adaptarlo al concept drift — mira cómo monitorear modelos de ML en producción con Evidently para detectar cuándo el drift realmente está ocurriendo.
• Modelos más sofisticados: LSTM Autoencoders para series temporales con patrones estacionales, o One-Class SVM para distribuciones no gaussianas.
• Kafka Streams / Faust: para lógica de procesamiento más compleja (ventanas de tiempo, joins entre topics) considera un framework de stream processing dedicado.
• Schema Registry: usa Avro o Protobuf con Confluent Schema Registry para validar el esquema de los mensajes y evolucionar los contratos entre productor y consumidor.

Conclusión

La combinación de Apache Kafka y scikit-learn permite construir sistemas de detección de anomalías en tiempo real con relativamente pocas líneas de código. Kafka se encarga del transporte confiable y escalable de eventos; el modelo de Isolation Forest se encarga de la detección. El desacoplamiento entre productor y consumidor garantiza que puedes escalar, mejorar o reemplazar cualquier parte sin afectar las demás.

El código completo con Docker Compose, productor, consumidor y entrenamiento está en GitHub.