Salta el contingut

Alertes i monitorització: enterar-te ABANS que el CEO pregunti

Totes les tècniques anteriors (pytest, Pandera, Great Expectations, contractes de dades) validen les dades en un moment concret. Però la qualitat de dades no és estàtica: un pipeline que funcionava perfectament fa tres mesos pot degradar-se gradualment perquè una API externa ha canviat silenciosament el seu format, perquè un origen de dades ha deixat d'actualitzar-se, o perquè el volum de dades ha caigut sense que ningú se n'adoni. Cal, doncs, un sistema de monitorització contínua que detecti aquesta degradació i alerti abans que algú de negoci detecti un número estrany en un dashboard.

Què monitoritzar

Què Pregunta que respon Exemple de mètrica
Frescor Les dades s'estan actualitzant amb la freqüència esperada? Temps transcorregut des de l'última càrrega
Volum Arriba el volum de dades esperat, ni més ni menys? Nombre de files carregades avui vs. mitjana dels últims 7 dies
Distribucions Els valors es mantenen dins del rang/forma habitual? Mitjana i desviació estàndard d'import avui vs. històric
Taxa d'errors de validació Quin percentatge de registres falla les regles de qualitat? % de registres rebutjats per Great Expectations en l'última execució
Schema drift Ha canviat l'estructura de les dades d'origen? Comparació de columnes/tipus rebuts vs. contracte de dades 
flowchart TB
    A["Pipeline\nen execució"] --> B{"Mètriques\ndins del llindar?"}
    B -->|"Sí"| C["Continua normal"]
    B -->|"No"| D["Genera alerta"]
    D --> E["Slack / Email\nequip de dades"]
    D --> F["Registre a l'historial\nde qualitat"]

    style A fill:#1d4ed8,color:#ffffff,stroke:#3b82f6
    style B fill:#b45309,color:#ffffff,stroke:#f59e0b
    style C fill:#166534,color:#ffffff,stroke:#22c55e
    style D fill:#7f1d1d,color:#ffffff,stroke:#ef4444
    style E fill:#7c3aed,color:#ffffff,stroke:#a78bfa
    style F fill:#7c3aed,color:#ffffff,stroke:#a78bfa

Llindars estàtics vs dinàmics

Un llindar estàtic és un valor fix definit manualment: "alerta si el volum de vendes diàries és menor de 500". És senzill d'implementar però té un problema evident: no s'adapta a patrons naturals com l'estacionalitat (un diumenge sempre té menys vendes que un divendres) ni al creixement del negoci amb el temps.

VOLUM_MINIM_DIARI = 500

def comprova_volum_estatic(n_registres_avui: int) -> bool:
    return n_registres_avui >= VOLUM_MINIM_DIARI

Un llindar dinàmic es calcula a partir de l'històric recent, i per tant s'adapta sol:

import pandas as pd

def comprova_volum_dinamic(
    historic_diari: pd.Series,   # nombre de registres per dia, últims N dies
    n_registres_avui: int,
    desviacions_permeses: float = 2.0,
) -> tuple[bool, str]:
    mitjana = historic_diari.mean()
    desviacio = historic_diari.std()

    llindar_inferior = mitjana - desviacions_permeses * desviacio
    llindar_superior = mitjana + desviacions_permeses * desviacio

    if n_registres_avui < llindar_inferior:
        return False, f"Volum molt per sota de l'habitual: {n_registres_avui} (esperat ≥ {llindar_inferior:.0f})"
    if n_registres_avui > llindar_superior:
        return False, f"Volum anòmalament alt: {n_registres_avui} (esperat ≤ {llindar_superior:.0f})"
    return True, "Volum dins de l'esperat"

Aquest patró (mitjana ± N desviacions estàndard) és un mètode senzill de detecció d'anomalies, adequat com a primera capa de monitorització abans de necessitar res més sofisticat com un model de sèries temporals.

Enviar alertes: Slack i email

import requests

def alerta_slack(webhook_url: str, missatge: str, critica: bool = False) -> None:
    icona = "🔴" if critica else "🟡"
    payload = {"text": f"{icona} *Alerta de qualitat de dades*\n{missatge}"}
    resposta = requests.post(webhook_url, json=payload, timeout=5)
    resposta.raise_for_status()

import smtplib
from email.message import EmailMessage

def alerta_email(destinataris: list[str], assumpte: str, cos: str) -> None:
    msg = EmailMessage()
    msg["Subject"] = f"[Qualitat de dades] {assumpte}"
    msg["From"] = "pipelines@empresa.cat"
    msg["To"] = ", ".join(destinataris)
    msg.set_content(cos)

    with smtplib.SMTP("smtp.empresa.cat", 587) as servidor:
        servidor.starttls()
        servidor.send_message(msg)

El soroll d'alertes és tan perillós com la manca d'alertes

Un sistema que envia 40 alertes al dia acaba sent ignorat ("alert fatigue"): l'equip deixa de llegir-les i les alertes importants es perden entre les irrellevants. És preferible tenir poques alertes ben calibrades (amb llindars dinàmics i un nivell de severitat clar) que moltes alertes sorolloses.

Quarantena: no aturar el pipeline sencer per un registre dolent

Un pipeline robust no s'atura necessàriament davant d'un registre invàlid puntual; sovint és preferible separar-lo en una zona de quarantena per revisar-lo després, i deixar continuar la resta del pipeline amb les dades vàlides:

import logging
import pandas as pd
import pandera as pa

def processa_amb_quarantena(
    df: pd.DataFrame,
    esquema: pa.DataFrameSchema,
    taula_quarantena: str,
    conn,
) -> pd.DataFrame:
    """Separa registres vàlids d'invàlids i guarda la quarantena a una taula a part."""
    try:
        df_valids = esquema.validate(df, lazy=True)
        return df_valids
    except pa.errors.SchemaErrors as e:
        invalids = e.failure_cases
        invalids["timestamp_error"] = pd.Timestamp.now()
        invalids.to_sql(taula_quarantena, conn, if_exists="append", index=False)

        pct_error = len(invalids) / len(df) * 100
        logging.warning(
            "Registres enviats a quarantena: %d de %d (%.1f%%)",
            len(invalids), len(df), pct_error,
        )

        if pct_error > 10:
            alerta_slack(
                webhook_url="https://hooks.slack.com/services/...",
                missatge=f"{pct_error:.1f}% de registres invàlids en aquesta càrrega. Revisió urgent.",
                critica=True,
            )

        # Retorna només els registres vàlids restants (aproximació simplificada)
        indexs_invalids = invalids["index"].unique() if "index" in invalids else []
        return df.drop(index=indexs_invalids, errors="ignore")

Aquest patró combina tot el que s'ha vist en aquest bloc: valida (Pandera/GE), quarantena els registres dolents en lloc de descartar-los silenciosament, monitoritza el percentatge d'error, i alerta només quan aquest percentatge supera un llindar rellevant.

Mòdul M5074 Sistemes de Big Data | Institut Sa Palomera (Blanes) | Curs CEIABD 2026-2027