Cómo añadir logging a un pipeline ETL: paso a paso

Un pipeline ETL que se ejecuta en silencio es un problema esperando a suceder: cuando algo falla en mitad de la noche, no sabes dónde ni por qué. Añadir logging a un pipeline ETL te da un registro claro de cada fase — extracción, transformación y carga — para detectar errores pronto, medir cuánto tarda y saber cuántas filas se han procesado. A diferencia de un simple print(), el logging te ofrece niveles de gravedad, fecha y hora automáticas y la opción de escribir en varios destinos a la vez. Vas a construir, paso a paso, un sistema sencillo de logging y monitorización en Python usando solo la biblioteca estándar.

Requisitos previos

• Python 3.9 o superior instalado.
• Un pipeline ETL, aunque sea pequeño (leer datos, transformar, escribir).
• Conocimientos básicos de funciones en Python.
• Un editor de código, como VS Code.

Paso 1: Configurar el logging básico

El módulo logging forma parte de la biblioteca estándar de Python, así que no hay nada que instalar. Empieza por definir el formato de los mensajes y el nivel mínimo que quieres registrar. El nivel INFO es ideal para seguir el progreso del día a día; reserva DEBUG para cuando de verdad estés investigando un problema.

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger("etl")
logger.info("Pipeline ETL iniciado")

Paso 2: Registrar cada fase del pipeline

Registra el inicio de cada fase y, sobre todo, el número de filas procesadas. Estos recuentos son la forma más sencilla de monitorización: si un día extraes 0 filas donde solías tener miles, el log delata el problema de inmediato. Usa el nivel INFO para el flujo normal y WARNING cuando algo es raro pero no detiene el pipeline.

def extract():
    logger.info("Extração: a ler dados da origem")
    rows = read_source()          # a tua função de leitura
    logger.info("Extração: %d linhas lidas", len(rows))
    return rows

def transform(rows):
    logger.info("Transformação: a aplicar regras")
    clean = [r for r in rows if r["valor"] is not None]
    logger.info("Transformação: %d válidas de %d", len(clean), len(rows))
    return clean

Paso 3: Guardar los logs en un fichero con rotación

Ver los mensajes en pantalla ayuda durante el desarrollo, pero en producción necesitas almacenarlos. El RotatingFileHandler escribe en un fichero y crea automáticamente uno nuevo cuando el actual alcanza el tamaño límite, evitando ficheros gigantes que llenan el disco.

from logging.handlers import RotatingFileHandler

file_handler = RotatingFileHandler(
    "etl.log", maxBytes=1_000_000, backupCount=5
)
file_handler.setFormatter(
    logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
)
logger.addHandler(file_handler)

Paso 4: Capturar y registrar errores

Envuelve cada fase en un bloque try/except. El método logger.exception() registra el mensaje y el traceback completo, lo que hace que analizar el error sea mucho más rápido. Vuelve a lanzar la excepción después para que el pipeline se detenga de forma controlada en lugar de seguir con datos incompletos. Así los errores nunca pasan desapercibidos y quedan guardados con todo su contexto.

def load(rows):
    try:
        logger.info("Carga: a gravar %d linhas", len(rows))
        write_target(rows)        # a tua função de escrita
        logger.info("Carga: concluída com sucesso")
    except Exception:
        logger.exception("Carga: falhou ao gravar no destino")
        raise

Paso 5: Medir cuánto tarda cada ejecución

La duración es una métrica de monitorización muy útil: un pipeline que suele tardar 2 minutos y de repente tarda 30 es una clara señal de alarma. Usa time.perf_counter() para medir y registrar el tiempo total de cada ejecución.

import time

def run_pipeline():
    start = time.perf_counter()
    logger.info("=== Execução iniciada ===")
    rows = transform(extract())
    load(rows)
    elapsed = time.perf_counter() - start
    logger.info("=== Execução concluída em %.1f s ===", elapsed)

if __name__ == "__main__":
    run_pipeline()

Comprobar el resultado

Ejecuta el pipeline y abre el fichero etl.log. Deberías ver una línea por cada fase, con fecha, hora, nivel y mensaje — incluyendo los recuentos de filas y la duración final. Para probar el manejo de errores, fuerza un fallo a propósito (por ejemplo, apunta la lectura a un fichero que no existe) y confirma que aparece una línea ERROR con el traceback. Después, filtra solo los problemas con un comando rápido:

grep ERROR etl.log

Conclusión

Con apenas unas líneas de código has pasado de un pipeline "caja negra" a un proceso observable: sabes qué se ejecuta, cuánto tarda y dónde falla. El siguiente paso natural es enviar estos logs a una herramienta central, como Azure Monitor o Application Insights, y crear alertas automáticas cada vez que aparezca un ERROR. ¿Qué métrica crees que es más importante vigilar en tu pipeline: el número de filas procesadas o el tiempo de ejecución?