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 sugirió

En tu 1:1 del miércoles, Elena te dijo algo que se te quedó grabado: "Lo que diferencia a un junior que crece rápido de uno que se estanca es que el primero automatiza las cosas que le molestan. Si algo te hizo perder tiempo esta semana, piensa cómo evitarlo la próxima." Dos cosas te hicieron perder tiempo:

01.Descubrir que los CSVs de María tenían problemas DESPUÉS de empezar a procesarlos. Si hubieras tenido un validador automático, habrías detectado los 3 formatos de fecha y los duplicados en 2 segundos.
02.El jueves tuviste suerte: el DAG de pedidos falló con un error y Airflow avisó a las 9:15. Pero un pipeline puede terminar en verde y no haber escrito ni una fila — un filtro que se lleva todo por delante, un fichero de origen vacío, una partición que se escribió en la carpeta de otro día. Eso no falla, así que no avisa. Y de eso te enteras cuando Jorge pregunta «oye, ¿no hay datos de ayer?».

Estas dos mejoras son pequeñas 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 automático de CSVs

La idea es simple: antes de procesar cualquier CSV que llegue de Marketing (o de cualquier fuente), ejecutas una función que verifica que el archivo cumple con lo esperado. Si no cumple, te dice exactamente QUÉ falla — antes de que pierdas una hora limpiando datos que ni siquiera están completos.

Un buen validador de CSVs verifica al menos tres cosas:

01.Estructura: las columnas esperadas existen y están en el orden correcto.
02.Tipos: los valores de cada columna son del tipo esperado (números donde debe haber números, fechas donde debe haber fechas).
03.Completitud: las columnas críticas 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 numéricos"
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 válida"
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 crítica '{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 patrón de "validar antes de procesar" se llama "fail fast". Es mejor descubrir que un archivo está mal en 2 segundos que después de 30 minutos de transformaciones. En equipos maduros, la validación es el PRIMER paso de cualquier pipeline — incluso antes de mover el archivo a raw.

### Parte 2: Detectar gaps en el pipeline

¿Cómo sabes si tu pipeline funcionó ayer? En FreshMart el pipeline diario ingesta datos de ventas y los deja en la tabla gold_ventas_diarias. Si un día 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 solución es una query de monitoreo que ejecutas cada mañana (o que automatizas con un cron job) y que detecta "huecos" — días donde debería 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 falló en el pipeline

En un equipo maduro, esta query se ejecuta automáticamente cada mañana a las 7:00 AM (antes de que llegue nadie). Si detecta gaps, envía una notificación a Slack con el mensaje: "ALERTA: no hay datos de ventas para el día X. Revisar DAG raw_to_gold_ventas." Así, cuando Jorge llega a las 9:00, el problema ya está 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 integración 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 ACCIÓN HUMANA. Si el sistema se puede recuperar solo, que se recupere — y deja un log. Si necesita intervención, alerta al canal con instrucciones claras de qué hacer.

### Por qué 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:

Validación 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 línea.
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 después de un incidente — impleméntalos ANTES. El mejor momento para poner un validador es cuando todo funciona bien. El peor momento es cuando ya perdiste datos y el CEO está preguntando qué pasó.

Regístrate para guardar tu progreso.