lección 10

Bonus -- Automatizar validaciones y detectar fallos

Dos tareas extra que complementan tu semana: un validador de CSVs para Marketing y una query que detecta gaps en el pipeline.

⏱ 30 min

### El contexto: dos mejoras que Elena te sugirio

En tu 1:1 del miercoles, Elena te dijo algo que se te quedo grabado: "Lo que diferencia a un junior que crece rapido de uno que se estanca es que el primero automatiza las cosas que le molestan. Si algo te hizo perder tiempo esta semana, piensa como evitarlo la proxima." Dos cosas te hicieron perder tiempo:

01.Descubrir que los CSVs de Maria tenian problemas DESPUES de empezar a procesarlos. Si hubieras tenido un validador automatico, habrias detectado los 3 formatos de fecha y los duplicados en 2 segundos.
02.No saber si el pipeline del dia anterior genero datos o no. Te enteraste porque Jorge pregunto "oye, no hay datos de ayer en gold" -- y ahi empezaste a investigar. Si hubiera una alerta automatica, lo habrias sabido antes que Jorge.

Estas dos mejoras son pequenas pero poderosas. Son exactamente el tipo de cosas que un junior proactivo hace en su primera semana -- y que impresionan a su tech lead.

### Parte 1: Validador automatico de CSVs

La idea es simple: antes de procesar cualquier CSV que llegue de Marketing (o de cualquier fuente), ejecutas una funcion que verifica que el archivo cumple con lo esperado. Si no cumple, te dice exactamente QUE falla -- antes de que pierdas una hora limpiando datos que ni siquiera estan completos.

Un buen validador de CSVs verifica al menos tres cosas:

01.Estructura: las columnas esperadas existen y estan en el orden correcto.
02.Tipos: los valores de cada columna son del tipo esperado (numeros donde debe haber numeros, fechas donde debe haber fechas).
03.Completitud: las columnas criticas no tienen valores nulos por encima de un umbral aceptable.

1# csv_validator.py -- Validador reutilizable para CSVs de Marketing
2import pandas as pd
3from typing import Dict, List, Optional
4
5
6class ValidationResult:
7    """Resultado de una validacion con errores acumulados."""
8
9    def __init__(self):
10        self.errors: List[str] = []
11        self.warnings: List[str] = []
12
13    @property
14    def is_valid(self) -> bool:
15        return len(self.errors) == 0
16
17    def add_error(self, msg: str):
18        self.errors.append(msg)
19
20    def add_warning(self, msg: str):
21        self.warnings.append(msg)
22
23    def summary(self) -> str:
24        lines = []
25        if self.is_valid:
26            lines.append("VALIDACION OK")
27        else:
28            lines.append(f"VALIDACION FALLIDA -- {len(self.errors)} error(es)")
29        for e in self.errors:
30            lines.append(f"  ERROR: {e}")
31        for w in self.warnings:
32            lines.append(f"  WARNING: {w}")
33        return "\n".join(lines)
34
35
36def validate_csv(
37    df: pd.DataFrame,
38    expected_columns: List[str],
39    column_types: Optional[Dict[str, str]] = None,
40    max_null_pct: float = 0.05,
41    critical_columns: Optional[List[str]] = None,
42) -> ValidationResult:
43    """
44    Valida un DataFrame contra un esquema esperado.
45
46    Args:
47        df: DataFrame cargado del CSV
48        expected_columns: lista de columnas que DEBEN existir
49        column_types: dict {columna: tipo_esperado} donde tipo es
50                      'numeric', 'datetime', 'string'
51        max_null_pct: porcentaje maximo de nulos aceptable (0.05 = 5%)
52        critical_columns: columnas que NO pueden tener ningun nulo
53
54    Returns:
55        ValidationResult con errores y warnings
56    """
57    result = ValidationResult()
58
59    # 1. Verificar columnas esperadas
60    missing = set(expected_columns) - set(df.columns)
61    extra = set(df.columns) - set(expected_columns)
62    if missing:
63        result.add_error(f"Columnas faltantes: {sorted(missing)}")
64    if extra:
65        result.add_warning(f"Columnas extra (no esperadas): {sorted(extra)}")
66
67    # 2. Verificar tipos de datos
68    if column_types:
69        for col, expected_type in column_types.items():
70            if col not in df.columns:
71                continue
72            if expected_type == 'numeric':
73                non_numeric = pd.to_numeric(df[col], errors='coerce').isna()
74                original_na = df[col].isna()
75                bad_values = non_numeric & ~original_na
76                if bad_values.sum() > 0:
77                    result.add_error(
78                        f"Columna '{col}': {bad_values.sum()} valores no numericos"
79                    )
80            elif expected_type == 'datetime':
81                non_date = pd.to_datetime(df[col], errors='coerce', dayfirst=True).isna()
82                original_na = df[col].isna()
83                bad_values = non_date & ~original_na
84                if bad_values.sum() > 0:
85                    result.add_error(
86                        f"Columna '{col}': {bad_values.sum()} valores no son fecha valida"
87                    )
88
89    # 3. Verificar nulos
90    if critical_columns:
91        for col in critical_columns:
92            if col not in df.columns:
93                continue
94            null_count = df[col].isna().sum()
95            if null_count > 0:
96                result.add_error(
97                    f"Columna critica '{col}' tiene {null_count} nulos"
98                )
99
100    # Verificar umbral general de nulos
101    for col in df.columns:
102        null_pct = df[col].isna().mean()
103        if null_pct > max_null_pct:
104            result.add_warning(
105                f"Columna '{col}': {null_pct:.1%} nulos (umbral: {max_null_pct:.1%})"
106            )
107
108    # 4. Verificar duplicados
109    dup_count = df.duplicated().sum()
110    if dup_count > 0:
111        result.add_warning(f"Filas duplicadas exactas: {dup_count}")
112
113    return result
114
115
116# Ejemplo de uso con los datos de Maria
117if __name__ == "__main__":
118    df = pd.read_csv("datos/campana_verano_ventas.csv")
119
120    result = validate_csv(
121        df=df,
122        expected_columns=[
123            'venta_id', 'cliente_id', 'producto_id',
124            'fecha_compra', 'cantidad', 'precio'
125        ],
126        column_types={
127            'cantidad': 'numeric',
128            'fecha_compra': 'datetime',
129        },
130        critical_columns=['venta_id', 'producto_id', 'fecha_compra'],
131        max_null_pct=0.03,
132    )
133
134    print(result.summary())

Un validador profesional que puedes reutilizar para cualquier CSV que llegue al equipo

Consejo de senior: este patron de "validar antes de procesar" se llama "fail fast". Es mejor descubrir que un archivo esta mal en 2 segundos que despues de 30 minutos de transformaciones. En equipos maduros, la validacion es el PRIMER paso de cualquier pipeline -- incluso antes de mover el archivo a raw.

### Parte 2: Detectar gaps en el pipeline

El segundo problema es mas sutil: como sabes si tu pipeline funciono ayer? En FreshMart el pipeline diario ingesta datos de ventas y los deja en la tabla gold_ventas_diarias. Si un dia el pipeline falla en silencio (sin error pero sin generar datos), nadie se entera hasta que un analista pregunta "oye, faltan datos de ayer".

La solucion es una query de monitoreo que ejecutas cada manana (o que automatizas con un cron job) y que detecta "huecos" -- dias donde deberia haber datos pero no los hay.

1-- detectar_gaps_pipeline.sql
2-- Encuentra dias sin datos en la tabla gold de ventas diarias
3-- Se asume que TODOS los dias deberian tener al menos 1 registro
4
5-- 1. Generar serie de fechas esperadas (ventana de 30 dias)
6-- Usamos una fecha de referencia FIJA porque el dataset es historico de 2024.
7WITH fechas_esperadas AS (
8    SELECT gs::date AS fecha
9    FROM generate_series(
10        DATE '2024-04-20' - INTERVAL '30 days',
11        DATE '2024-04-20' - INTERVAL '1 day',
12        INTERVAL '1 day'
13    ) AS t(gs)
14),
15
16-- 2. Obtener las fechas que SI tienen datos
17fechas_con_datos AS (
18    SELECT DISTINCT fecha_venta AS fecha
19    FROM gold_ventas_diarias
20    WHERE fecha_venta >= DATE '2024-04-20' - INTERVAL '30 days'
21),
22
23-- 3. Encontrar los gaps (fechas sin datos)
24gaps AS (
25    SELECT
26        fe.fecha,
27        CASE
28            WHEN fd.fecha IS NULL THEN 'SIN DATOS'
29            ELSE 'OK'
30        END AS estado
31    FROM fechas_esperadas fe
32    LEFT JOIN fechas_con_datos fd ON fe.fecha = fd.fecha
33    WHERE fd.fecha IS NULL
34)
35
36SELECT
37    fecha,
38    estado,
39    fecha - LAG(fecha) OVER (ORDER BY fecha) AS dias_desde_ultimo_gap
40FROM gaps
41ORDER BY fecha DESC;
42
43-- Si esta query devuelve filas, hay dias sin datos
44-- En un sistema de alertas, esto dispararia una notificacion

Query de monitoreo -- si devuelve filas, algo fallo en el pipeline

En un equipo maduro, esta query se ejecuta automaticamente cada manana a las 7:00 AM (antes de que llegue nadie). Si detecta gaps, envia una notificacion a Slack con el mensaje: "ALERTA: no hay datos de ventas para el dia X. Revisar DAG raw_to_gold_ventas." Asi, cuando Jorge llega a las 9:00, el problema ya esta diagnosticado.

1# alertas_slack.py -- Ejemplo de como se integraria con Slack
2import requests
3from datetime import date
4
5
6def enviar_alerta_slack(webhook_url: str, gaps: list[date]):
7    """
8    Envia una alerta a Slack cuando hay dias sin datos.
9    En un entorno real, el webhook_url seria una variable de entorno.
10    """
11    if not gaps:
12        return  # No hay gaps, no alertar
13
14    fechas_str = ", ".join(f.strftime("%Y-%m-%d") for f in gaps)
15    mensaje = {
16        "text": (
17            f":warning: *ALERTA: Pipeline sin datos*\n"
18            f"Dias afectados: {fechas_str}\n"
19            f"Total gaps: {len(gaps)} dia(s)\n"
20            f"Accion: revisar DAG 'raw_to_gold_ventas' en Airflow"
21        )
22    }
23
24    response = requests.post(webhook_url, json=mensaje)
25    if response.status_code == 200:
26        print(f"Alerta enviada a Slack: {len(gaps)} gap(s) detectado(s)")
27    else:
28        print(f"Error enviando alerta: {response.status_code}")
29
30
31# En produccion esto iria en un DAG de Airflow que corre a las 7:00 AM
32# O en un cron job: 0 7 * * * python alertas_slack.py

La integracion con Slack es sorprendentemente simple -- un POST a un webhook

Consejo de senior: las alertas son un arma de doble filo. Si alertas por TODO, la gente las ignora (alert fatigue). Si alertas por NADA, te enteras tarde. La regla de oro: alerta solo cuando se requiere ACCION HUMANA. Si el sistema se puede recuperar solo, que se recupere -- y deja un log. Si necesita intervencion, alerta al canal con instrucciones claras de que hacer.

### Por que estas dos cosas importan tanto

Validar CSVs y detectar gaps parecen tareas menores. Pero juntas representan los dos pilares de la confianza en un sistema de datos:

Validacion de entrada: te aseguras de que lo que entra al sistema es correcto ANTES de procesarlo. Es el portero del club.
Monitoreo de salida: te aseguras de que lo que sale del sistema es completo y a tiempo. Es el control de calidad al final de la linea.
Sin estos dos controles, tu pipeline es una caja negra: entra algo, sale algo, pero nadie sabe si lo que sale es correcto o completo.
Con estos dos controles, puedes dormir tranquilo. Si algo falla, te enteras antes que el negocio.

No implementes estos controles despues de un incidente -- implementalos ANTES. El mejor momento para poner un validador es cuando todo funciona bien. El peor momento es cuando ya perdiste datos y el CEO esta preguntando que paso.

## ejercicios

[01]

Validador de CSVs de Marketing

Implementa la funcion validate_csv que verifica: 1) que las columnas esperadas existen, 2) que los valores numericos son realmente numeros, 3) que las columnas criticas no tienen nulos. La funcion recibe un DataFrame y un esquema, y devuelve un objeto con los errores encontrados.

Cargando editor...

[02]

Detectar dias sin datos en el pipeline

Escribe una query SQL que detecte "gaps" -- dias donde deberia haber datos de ventas pero no los hay. La query debe generar una serie de los ultimos 30 dias, cruzarla con los dias que SI tienen datos, y devolver los dias faltantes con una columna de estado.

Cargando editor...

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...