Saltar al contenido
~/bigdatastack

lección 7

Alertas y monitorización: enterarte ANTES de que el CEO pregunte

Qué monitorizar (frescura, volumen, distribuciones), cómo alertar (Slack, email), umbrales estáticos y dinámicos.

⏱ 50 min

### El pipeline perfecto que nadie vigila

Tienes validaciones. Tienes contratos. Tienes Great Expectations ejecutándose cada día. Genial. Pero... ¿quién VE los resultados? Si tu suite de GE falla a las 3 de la mañana un sábado y el resultado queda en un log que nadie lee hasta el lunes, ¿de qué sirve? De nada. Una validación sin alerta es como una alarma de incendios sin sirena — detecta el fuego pero nadie se entera.

La monitorización de datos es el último eslabón de la cadena de calidad. Completa el ciclo: defines expectativas (contrato) → verificas que se cumplen (validación) → alertas cuando NO se cumplen (monitorización). Sin los tres eslabones, la cadena se rompe.

Analogía: imagina un hospital. Los pacientes están monitorizados con máquinas que miden pulso, presión, oxígeno. Si algo se sale del rango normal, la máquina PITA. No espera a que una enfermera pase por ahí y mire la pantalla — avisa activamente. Tus datos necesitan lo mismo: monitorización activa con alertas inmediatas.

### ¿Qué monitorizar? Las 4 señales vitales de los datos

  1. 01.FRESCURA: ¿los datos están actualizados? Si tu pipeline debería ejecutarse cada hora, verifica que realmente lo hizo. Si los datos tienen más de X horas, alerta.
  2. 02.VOLUMEN: ¿llegó la cantidad esperada de datos? Si normalmente recibes 10.000 pedidos diarios y hoy llegaron 200, algo está mal. Si llegaron 500.000, también.
  3. 03.DISTRIBUCIÓN: ¿los valores se comportan "normalmente"? Si el importe medio siempre es ~50€ y hoy es 5.000€, hay un problema. Si el % de nulos sube del 1% habitual al 30%, hay un problema.
  4. 04.SCHEMA: ¿la estructura cambió? Si apareció una columna nueva o desapareció una existente, alerta (posible breaking change del upstream).
Una alerta sin contexto es ruido. Una alerta con contexto es acción.

### Umbrales estáticos vs dinámicos

Un umbral estático es simple: "si hay más del 5% de nulos, alerta". Funciona, pero tiene un problema: no se adapta. Si un dataset normalmente tiene 0.1% de nulos y sube a 3%, el umbral estático no se dispara (está por debajo del 5%), pero claramente algo cambió. Los umbrales dinámicos se basan en el HISTORIAL: "si el % de nulos es 3 desviaciones estándar mayor que la media de los últimos 30 días, alerta".

1import pandas as pd
2import numpy as np
3from datetime import datetime, timedelta
4

5class MonitorDatos:
6    """Monitor de calidad con umbrales estáticos y dinámicos."""
7    
8    def __init__(self, nombre_dataset: str):
9        self.nombre = nombre_dataset
10        self.historial: list[dict] = []
11    
12    def registrar_metricas(self, metricas: dict):
13        """Registra las métricas del día (simula historial)."""
14        metricas['timestamp'] = datetime.now()
15        self.historial.append(metricas)
16    
17    def check_estatico(self, metrica: str, valor: float, umbral: float) -> dict:
18        """Check simple: ¿el valor supera el umbral fijo?"""
19        dispara = valor > umbral
20        return {
21            'tipo': 'estático',
22            'metrica': metrica,
23            'valor': valor,
24            'umbral': umbral,
25            'alerta': dispara,
26            'mensaje': f"{metrica} = {valor:.2f} ({'[CRITICO] > ' if dispara else '[OK] <= '}{umbral})",
27        }
28    
29    def check_dinamico(self, metrica: str, valor_actual: float, 
30                        historial_valores: list[float], sigma: float = 3.0) -> dict:
31        """Check dinámico: ¿el valor se desvía del comportamiento histórico?"""
32        if len(historial_valores) < 7:
33            return {'tipo': 'dinámico', 'metrica': metrica, 'alerta': False,
34                    'mensaje': 'Historial insuficiente (< 7 días)'}
35        
36        media = np.mean(historial_valores)
37        std = np.std(historial_valores)
38        
39        if std == 0:
40            limite_superior = media * 1.5
41            limite_inferior = media * 0.5
42        else:
43            limite_superior = media + sigma * std
44            limite_inferior = media - sigma * std
45        
46        fuera = valor_actual > limite_superior or valor_actual < limite_inferior
47        
48        return {
49            'tipo': 'dinámico',
50            'metrica': metrica,
51            'valor': valor_actual,
52            'media_historica': round(media, 2),
53            'rango_normal': f"[{limite_inferior:.1f}, {limite_superior:.1f}]",
54            'alerta': fuera,
55            'mensaje': f"{metrica} = {valor_actual:.1f} vs media {media:.1f} "
56                       f"{'[CRITICO] ANOMALÍA' if fuera else '[OK] Normal'}",
57        }
58

59# Ejemplo: monitorizar volumen diario
60monitor = MonitorDatos("pedidos_diarios")
61

62# Historial de los últimos 14 días (volumen normal: ~10000 ± 1000)
63historial_volumen = [9800, 10200, 10100, 9900, 10500, 10300, 9700, 
64                     10100, 10400, 9600, 10200, 10000, 10300, 9800]
65

66# Hoy: solo llegaron 3200 pedidos (¡anomalía!)
67volumen_hoy = 3200
68

69# Check estático (umbral fijo: mínimo 5000)
70resultado_estatico = monitor.check_estatico("volumen", volumen_hoy, 5000)
71print(f"Estático: {resultado_estatico['mensaje']}")
72

73# Check dinámico (basado en historial)
74resultado_dinamico = monitor.check_dinamico("volumen", volumen_hoy, historial_volumen)
75print(f"Dinámico: {resultado_dinamico['mensaje']}")
76print(f"  Rango normal: {resultado_dinamico.get('rango_normal', 'N/A')}")

El umbral dinámico detecta la anomalía aunque el estático no se dispare — se adapta al comportamiento real

### Implementar alertas: Slack, email y más

En producción real, las alertas se envían a canales donde el equipo las vea inmediatamente. Los más comunes son: Slack (un canal #data-alerts), email (al equipo on-call), PagerDuty (para incidentes críticos a las 3 AM). Aquí vamos a implementar un sistema de alertas SIMULADO que muestra la arquitectura correcta — en producción solo cambiarías el "enviar" por la integración real.

1from dataclasses import dataclass
2from datetime import datetime
3from enum import Enum
4

5class Severidad(Enum):
6    INFO = "info"
7    WARNING = "warning" 
8    CRITICAL = "critical"
9

10@dataclass
11class Alerta:
12    """Una alerta de calidad de datos."""
13    dataset: str
14    metrica: str
15    mensaje: str
16    severidad: Severidad
17    valor_actual: float
18    valor_esperado: float
19    timestamp: datetime = None
20    
21    def __post_init__(self):
22        if self.timestamp is None:
23            self.timestamp = datetime.now()
24

25class SistemaAlertas:
26    """Sistema de alertas multi-canal."""
27    
28    def __init__(self):
29        self.alertas_enviadas: list[Alerta] = []
30    
31    def enviar(self, alerta: Alerta):
32        """Enruta la alerta al canal apropiado según severidad."""
33        self.alertas_enviadas.append(alerta)
34        
35        if alerta.severidad == Severidad.CRITICAL:
36            self._enviar_slack(alerta)
37            self._enviar_email(alerta)
38            self._enviar_pagerduty(alerta)
39        elif alerta.severidad == Severidad.WARNING:
40            self._enviar_slack(alerta)
41        else:
42            self._log(alerta)
43    
44    def _enviar_slack(self, alerta: Alerta):
45        """Simula envío a Slack (en producción: webhook o SDK)."""
46        emoji = "[CRITICO]" if alerta.severidad == Severidad.CRITICAL else "[AVISO]"
47        print(f"[SLACK #data-alerts] {emoji} {alerta.dataset}: {alerta.mensaje}")
48        print(f"  Valor: {alerta.valor_actual} (esperado: {alerta.valor_esperado})")
49        # En producción:
50        # requests.post(SLACK_WEBHOOK_URL, json={"text": mensaje})
51    
52    def _enviar_email(self, alerta: Alerta):
53        """Simula envío de email (en producción: SMTP o SES)."""
54        print(f"[EMAIL → data-oncall@empresa.com] CRÍTICO: {alerta.dataset}")
55        print(f"  {alerta.mensaje}")
56        # En producción:
57        # import smtplib / boto3.client('ses').send_email(...)
58    
59    def _enviar_pagerduty(self, alerta: Alerta):
60        """Simula envío a PagerDuty (en producción: API)."""
61        print(f"[PAGERDUTY] Incidente creado: {alerta.dataset} — {alerta.metrica}")
62        # En producción:
63        # requests.post("https://events.pagerduty.com/v2/enqueue", json=payload)
64    
65    def _log(self, alerta: Alerta):
66        """Solo log para alertas informativas."""
67        print(f"[LOG] {alerta.dataset}: {alerta.mensaje}")
68

69# Uso
70alertas = SistemaAlertas()
71

72# Alerta crítica: pipeline no se ejecutó
73alertas.enviar(Alerta(
74    dataset="pedidos_diarios",
75    metrica="frescura",
76    mensaje="Datos no actualizados en 6 horas (SLA: 1h)",
77    severidad=Severidad.CRITICAL,
78    valor_actual=6.0,
79    valor_esperado=1.0,
80))
81

82print()
83

84# Alerta warning: volumen bajo
85alertas.enviar(Alerta(
86    dataset="pedidos_diarios",
87    metrica="volumen",
88    mensaje="Volumen 40% menor que la media histórica",
89    severidad=Severidad.WARNING,
90    valor_actual=6000,
91    valor_esperado=10000,
92))

La severidad determina el canal: INFO → log, WARNING → Slack, CRITICAL → Slack + email + PagerDuty

Consejo de senior sobre alertas: la regla de oro es "si una alerta no requiere ACCIÓN, no debería ser una alerta". Las alertas informativas que nadie mira generan "fatiga de alertas" — el equipo deja de prestarles atención y cuando llega una real, la ignoran. Menos alertas, más accionables = mejor.

Error fatal que he visto múltiples veces: enviar TODAS las alertas al mismo canal con la misma severidad. Si el canal de Slack tiene 50 mensajes diarios de "todo OK" mezclados con 1 mensaje de "pipeline roto", nadie va a ver el importante. Separa los canales por severidad y SÍ SOLO envía al canal lo que requiere atención.

### Anatomía de una buena alerta

Una alerta sin contexto es inútil. "Pipeline falló" no le dice nada al ingeniero que la recibe a las 3 AM. Una buena alerta incluye:

  1. 01.QUÉ falló: nombre del dataset y métrica específica
  2. 02.CUÁNDO: timestamp exacto del fallo
  3. 03.CUÁNTO se desvía: valor actual vs esperado (no solo "está mal" sino "es un 40% menor")
  4. 04.DESDE CUÁNDO: ¿es nuevo o lleva horas/días?
  5. 05.IMPACTO: ¿a quién afecta? (dashboard X, informe Y, equipo Z)
  6. 06.RUNBOOK: link a documentación de cómo investigar/resolver este tipo de fallo

Lo que le diría a mi yo de hace 5 años: crea un runbook para CADA tipo de alerta. Un runbook es un documento que dice: "si recibes esta alerta, haz estos pasos: 1) verifica X, 2) mira el log Y, 3) si es Z reinicia el pipeline, 4) si no se resuelve escala a fulano". A las 3 AM no quieres estar pensando desde cero.

Con validaciones, contratos y alertas tienes el stack completo de calidad de datos. En la próxima (y última) lección lo vamos a juntar TODO en un proyecto real: un pipeline completo con validación, alertas y reporte de calidad.

Regístrate para guardar tu progreso.

## comentarios

Reporta erratas, ayuda a otros o comparte tu opinión. Sé constructivo.

Inicia sesión para comentar y responder.

cargando comentarios...