lección 6

Logging y manejo de errores en producción

Cuando tu pipeline falla a las 3AM, los logs son tu única pista. Aprende a dejar rastros útiles.

⏱ 45 min

### El pipeline invisible

Tu pipeline funciona. Corre cada noche a las 2AM mientras duermes. Nadie lo mira, nadie lo vigila. Hasta que un martes a las 7AM, el equipo de BI reporta que la tabla de ventas no se actualizó. ¿Qué pasó? ¿Falló la lectura? ¿La API devolvió un error? ¿Se llenó el disco? Sin logs, no tienes NI IDEA. Con logs buenos, sabes exactamente qué falló, cuándo, y por qué en menos de 2 minutos.

La analogía: los logs son como la caja negra de un avión. Cuando todo va bien, nadie los mira. Cuando algo explota, son lo ÚNICO que tienes para reconstruir qué pasó. Un pipeline sin logs es un avión sin caja negra: vuela bien hasta que no vuela, y entonces nadie sabe qué pasó.

### print() NO es logging

Sé lo que estás pensando: "pero yo pongo print() en mi código para ver qué pasa". El problema de print: no tiene timestamp (¿cuándo pasó?), no tiene niveles (¿es un error o solo información?), no se puede redirigir a un archivo, no se puede filtrar, y se pierde cuando cierras la terminal. Para un script de 10 líneas que ejecutas a mano, print vale. Para un pipeline de producción, necesitas logging de verdad.

1import logging
2
3# Configuración básica (haz esto UNA vez al inicio del script)
4logging.basicConfig(
5    level=logging.INFO,
6    format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
7    datefmt='%Y-%m-%d %H:%M:%S'
8)
9
10logger = logging.getLogger('mi_pipeline')
11
12# Los 5 niveles de logging
13logger.debug("Entrando en la función X con parámetro Y")    # Para debugging detallado
14logger.info("Procesadas 15,000 filas correctamente")          # Progreso normal
15logger.warning("El archivo tiene 3 columnas menos de lo esperado")  # Algo raro pero no fatal
16logger.error("No se pudo conectar a la base de datos")        # Algo falló
17logger.critical("Pipeline abortado: datos corruptos detectados")  # Todo roto

Los 5 niveles: debug < info < warning < error < critical

Consejo de senior: pon logs desde el DÍA 1. No esperes a que algo falle. El coste de añadir un logger.info() es cero. El coste de no tenerlo cuando necesitas diagnosticar un error a las 3AM es enorme. Lo que le diría a mi yo de hace 5 años: "pon un log en cada entrada de función, cada lectura de archivo y cada escritura a base de datos".

### Logging a archivo: logs que sobreviven al reinicio

1import logging
2
3# Configuración con salida a archivo Y a consola
4logging.basicConfig(
5    level=logging.INFO,
6    format='%(asctime)s | %(levelname)-8s | %(message)s',
7    datefmt='%Y-%m-%d %H:%M:%S',
8    handlers=[
9        logging.FileHandler('pipeline.log'),      # Guarda en archivo
10        logging.StreamHandler()                     # También muestra en consola
11    ]
12)
13
14logger = logging.getLogger('pipeline_ventas')
15
16# Ahora todos los logs van a pipeline.log Y a la consola
17logger.info("Inicio del pipeline de ventas")
18logger.info("Conectando a base de datos...")

FileHandler guarda logs en disco: sobreviven reinicios y cierres

En producción los archivos de log crecen sin fin. RotatingFileHandler (de logging.handlers) los parte automáticamente: cuando el archivo llega a un tamaño (por ejemplo 5 MB), lo renombra a pipeline.log.1 y empieza uno nuevo. TimedRotatingFileHandler hace lo mismo pero por tiempo (uno al día, uno a la semana). Sin rotación, un pipeline que corre 24/7 llena el disco en semanas.

### Manejo de errores: try/except profesional

Ya conoces try/except de la Skill 3. Ahora vamos a usarlo como un profesional: con logging, con tipos específicos de error, y con la lógica de "qué hago cuando algo falla" que necesita un pipeline de producción.

1import logging
2import pandas as pd
3from sqlalchemy import create_engine
4
5logger = logging.getLogger('pipeline')
6
7def extraer_datos(query: str, engine) -> pd.DataFrame:
8    """Extrae datos de la DB con manejo de errores profesional."""
9    try:
10        logger.info(f"Ejecutando query: {query[:50]}...")
11        df = pd.read_sql(query, engine)
12        logger.info(f"Extraídas {len(df)} filas correctamente")
13        return df
14    
15    except ConnectionError as e:
16        logger.error(f"No se pudo conectar a la base de datos: {e}")
17        raise  # Re-lanzar para que el caller decida qué hacer
18    
19    except pd.errors.DatabaseError as e:
20        logger.error(f"Error SQL: {e}")
21        raise
22    
23    except Exception as e:
24        logger.critical(f"Error inesperado en extracción: {type(e).__name__}: {e}")
25        raise

Try/except profesional: captura específica, log del error, y re-lanzar

NUNCA hagas except: pass (capturar y silenciar). Es la forma más rápida de tener bugs invisibles. Si capturas un error, al mínimo loguéalo. Mejor aún: loguéalo Y re-lánzalo (raise) para que el nivel superior decida qué hacer.

### Patrón: pipeline con logging completo

Este es el patrón que usarás en todos tus pipelines. Cada paso está logueado: inicio, progreso, resultado y errores. Cuando algo falle a las 3AM, abrirás el log y en 30 segundos sabrás exactamente dónde y por qué.

1import logging
2import pandas as pd
3import time
4
5logging.basicConfig(
6    level=logging.INFO,
7    format='%(asctime)s | %(levelname)-8s | %(message)s',
8    datefmt='%Y-%m-%d %H:%M:%S',
9    handlers=[logging.FileHandler('pipeline.log'), logging.StreamHandler()]
10)
11logger = logging.getLogger('pipeline_ventas')
12
13def ejecutar_pipeline():
14    """Pipeline completo con logging profesional."""
15    inicio = time.time()
16    logger.info("=" * 50)
17    logger.info("INICIO: Pipeline de ventas diarias")
18    logger.info("=" * 50)
19    
20    try:
21        # Paso 1: Extracción
22        logger.info("[1/3] Extrayendo datos de la API...")
23        df = extraer_de_api()
24        logger.info(f"[1/3] Extraídas {len(df)} filas en {time.time()-inicio:.1f}s")
25        
26        # Paso 2: Limpieza
27        logger.info("[2/3] Limpiando datos...")
28        filas_antes = len(df)
29        df = limpiar(df)
30        filas_despues = len(df)
31        logger.info(f"[2/3] Limpieza: {filas_antes} → {filas_despues} filas ({filas_antes - filas_despues} eliminadas)")
32        
33        if filas_despues < filas_antes * 0.5:
34            logger.warning(f"Se eliminó más del 50% de datos. Revisar fuente.")
35        
36        # Paso 3: Carga
37        logger.info("[3/3] Cargando en base de datos...")
38        cargar_en_db(df)
39        logger.info(f"[3/3] Cargadas {len(df)} filas correctamente")
40        
41        duracion = time.time() - inicio
42        logger.info(f"ÉXITO: Pipeline completado en {duracion:.1f}s")
43        
44    except Exception as e:
45        logger.critical(f"FALLO: Pipeline abortado tras {time.time()-inicio:.1f}s")
46        logger.critical(f"Error: {type(e).__name__}: {e}")
47        raise
48
49if __name__ == '__main__':
50    ejecutar_pipeline()

Pipeline con logging completo: cada paso documentado, tiempos medidos, alertas de anomalías

### Qué loguear y qué NO loguear

SÍ: Inicio y fin de cada paso con duración
SÍ: Número de filas procesadas/eliminadas/insertadas
SÍ: Errores con contexto (qué archivo, qué tabla, qué query)
SÍ: Anomalías y warnings (más nulos de lo normal, archivo más pequeño de lo esperado)
NO: Contraseñas, tokens, API keys (NUNCA loguees secretos)
NO: Datos personales (emails, nombres) en texto claro
NO: Cada fila individual (millones de logs = ilegible e inútil)

### Custom exceptions: errores con significado

En pipelines complejos, Exception genérica no dice mucho. Crea excepciones propias que describan el problema de negocio:

1class PipelineError(Exception):
2    """Error base del pipeline."""
3    pass
4
5class ExtraccionError(PipelineError):
6    """La fuente de datos no respondió o devolvió datos inválidos."""
7    pass
8
9class ValidacionError(PipelineError):
10    """Los datos no pasan las reglas de calidad."""
11    pass
12
13class CargaError(PipelineError):
14    """Error al escribir en el destino."""
15    pass
16
17# Uso
18def validar_datos(df):
19    if df.empty:
20        raise ValidacionError("DataFrame vacío: la fuente no devolvió datos")
21    if df['importe'].isnull().sum() > len(df) * 0.1:
22        raise ValidacionError(f"Más del 10% de importes son nulos ({df['importe'].isnull().sum()}/{len(df)})")

Excepciones personalizadas: errores que cuentan una historia

Las excepciones personalizadas tienen dos ventajas: 1) El nombre describe el problema (ValidacionError dice más que Exception), 2) Puedes capturarlas selectivamente: except ExtraccionError → reintentar, except ValidacionError → alertar y no cargar datos sucios.

### pytest: comprobar que tu código hace lo que dice

pytest es la herramienta estándar en Python para escribir tests automáticos. Un test es una función que empieza por test_ y que comprueba algo concreto: "si le paso un DataFrame vacío, ¿lanza ValidacionError?". Si la condición se cumple, el test pasa (verde); si no, falla (rojo con el motivo exacto). Lo instalas con pip install pytest y lo ejecutas con pytest archivo.py -v. En el ejercicio siguiente lo usarás para verificar que tus excepciones se lanzan cuando deben.

1# Instalar pytest (una vez)
2pip install pytest
3
4# Ejecutar los tests de un archivo
5pytest mi_archivo.py -v
6
7# Ejemplo de salida:
8# test_valida_df_vacio PASSED
9# test_valida_importes_negativos PASSED
10# 2 passed in 0.03s

pytest se instala con pip y se lanza desde terminal

La función pytest.raises(ErrorEsperado, match="texto") es el patrón para comprobar que un error SÍ se lanza. Si la función no lanza nada, el test falla. Si lanza otro error, también falla. Solo pasa si lanza exactamente el error esperado con un mensaje que contiene "texto".

## ejercicios

[01]

Añadir logging profesional a un pipeline

Tienes un pipeline sin logs. Añade logging completo: configuración con FileHandler, log de inicio/fin de cada paso, medición de tiempo total, y manejo de errores con log del error.

Cargando editor...

[02]

Crear excepciones de validación de datos

Crea una clase ValidacionError y una función validar_ventas(df) que lance errores descriptivos si: el DataFrame está vacío, tiene importes negativos, o tiene más del 10% de nulos en "importe".

Cargando editor...

[03]

Detectar y loguear anomalías

Escribe una función verificar_calidad(df) que loguee warnings si: el número de filas es menor a 100, hay más del 5% de nulos en cualquier columna, o hay outliers extremos en importe (usando IQR, como en la lección de limpieza).

💡 Resultado esperado

WARNING  | Solo 11 filas (esperado >= 100). ¿Datos incompletos?
WARNING  | Columna 'importe' tiene 9.1% nulos (umbral: 5%)
WARNING  | 1 importes por encima de 166 (IQR). Revisar
INFO     | Verificación completada: 11 filas, 2 columnas

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