Cómo manejar errores y reintentos en un pipeline ETL en Python

João Barros 12 de July de 2026 6 min de lectura

Un pipeline ETL que funciona perfectamente en tu portátil se rompe en producción por razones de lo más banales: la API responde 429, la base de datos cierra la conexión a mitad, un fichero llega con una fila corrupta. Manejar errores y reintentos en un pipeline ETL es lo que separa un script frágil de un proceso en el que se puede confiar. El objetivo aquí está claro: distinguir un fallo temporal de un error real, reintentar solo cuando tiene sentido y no perder datos en silencio.

Requisitos previos

Python 3.9 o superior instalado.
Las librerías requests y sqlalchemy (pip install requests sqlalchemy).
Un pipeline ETL sencillo ya en funcionamiento (extraer de una API, transformar, cargar en una base de datos).
Nociones básicas de try/except en Python.

Paso 1: Separar los errores transitorios de los permanentes

No todos los errores merecen otro intento. Un timeout o un HTTP 503 son transitorios: dentro de diez segundos la petición probablemente funcione. En cambio, un HTTP 401 (credenciales incorrectas) o un KeyError en una transformación son permanentes: reintentar mil veces devuelve siempre el mismo error y solo retrasa el pipeline.

Empieza por escribir esa regla en código, para dejar de adivinar:

import requests

HTTP_RETRY = {408, 429, 500, 502, 503, 504}

def e_transitorio(erro):
    if isinstance(erro, requests.exceptions.HTTPError):
        resposta = erro.response
        return resposta is not None and resposta.status_code in HTTP_RETRY
    return isinstance(erro, (requests.exceptions.ConnectionError,
                             requests.exceptions.Timeout))

Paso 2: Crear una función de reintento con backoff exponencial

Reintentar de inmediato, cinco veces seguidas, es la mejor forma de molestar a una API que ya está sobrecargada. La buena práctica es el backoff exponencial: esperar 1s, 2s, 4s, 8s... y añadir algo de aleatoriedad (jitter) para que varios procesos no reintenten todos en el mismo segundo.

import logging
import random
import time

log = logging.getLogger("etl")

def com_retry(funcao, tentativas=5, base=1.0, maximo=30.0):
    for tentativa in range(1, tentativas + 1):
        try:
            return funcao()
        except Exception as erro:
            if tentativa == tentativas or not e_transitorio(erro):
                raise
            espera = min(maximo, base * 2 ** (tentativa - 1))
            espera += random.uniform(0, espera * 0.1)
            log.warning("Tentativa %s falhou (%s). Nova tentativa em %.1fs",
                        tentativa, erro, espera)
            time.sleep(espera)

Fíjate en dos detalles importantes: si el error no es transitorio, el raise ocurre ya en el primer intento; y en el último intento el error también se propaga, en lugar de quedar oculto.

Paso 3: Aplicar el reintento a la extracción

La extracción es donde el pipeline toca el mundo exterior, así que ahí es donde los reintentos más valen. Envuelve la llamada HTTP en una función sin argumentos y pásala a com_retry:

URL = "https://api.exemplo.com/v1/encomendas"

def extrair_pagina(pagina):
    def pedido():
        resposta = requests.get(URL, params={"page": pagina}, timeout=30)
        resposta.raise_for_status()
        return resposta.json()["data"]
    return com_retry(pedido)

Define siempre el timeout. Sin él, una petición colgada bloquea el pipeline durante horas y ningún reintento llega a producirse. Si la API devuelve la cabecera Retry-After, respeta ese valor en lugar del backoff calculado.

Paso 4: Aislar las filas erróneas en cuarentena

En la transformación el problema es otro: una sola fila cuyo campo valor dice "N/A" no debe tumbar un lote de 200.000 registros. En vez de reventar, separa lo válido de lo inválido y guarda los rechazados en una dead-letter queue (un fichero o una tabla de cuarentena) junto con el motivo.

import json

def transformar(linhas):
    validas, quarentena = [], []
    for linha in linhas:
        try:
            validas.append({
                "id": int(linha["id"]),
                "email": linha["email"].strip().lower(),
                "valor": float(linha["valor"]),
            })
        except (KeyError, TypeError, ValueError, AttributeError) as erro:
            quarentena.append({"linha": linha, "erro": str(erro)})
    return validas, quarentena

def guardar_quarentena(registos, caminho="quarentena.jsonl"):
    with open(caminho, "a", encoding="utf-8") as f:
        for registo in registos:
            f.write(json.dumps(registo, ensure_ascii=False) + "\n")

Una regla sana: si la cuarentena supera un umbral que definas (por ejemplo el 5% del lote), el pipeline debe fallar. Muchas filas malas de repente no es un accidente, es un contrato de datos que ha cambiado.

Paso 5: Fallar de forma controlada y visible

La carga debe ser transaccional: o entra todo, o no entra nada. Y cuando el pipeline falla de verdad, tiene que fallar en voz alta: con un código de salida distinto de cero, para que el orquestador (Airflow, Azure Data Factory, cron) sepa que algo ha ido mal.

import sys

def executar():
    linhas = extrair_pagina(1)
    validas, quarentena = transformar(linhas)

    if quarentena:
        guardar_quarentena(quarentena)
    if len(quarentena) > 0.05 * max(len(linhas), 1):
        raise ValueError(f"Demasiadas linhas invalidas: {len(quarentena)}")

    with motor.begin() as ligacao:   # commit al final, rollback si hay error
        carregar(ligacao, validas)

    log.info("ETL concluido: %s carregadas, %s em quarentena",
             len(validas), len(quarentena))

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    try:
        executar()
    except Exception:
        log.exception("ETL falhou")
        sys.exit(1)

Verificar el resultado

No esperes a producción para descubrir si tu manejo de errores funciona. Pruébalo a propósito:

Apunta URL a un endpoint que devuelva 503 y confirma en los logs que las esperas crecen (1s, 2s, 4s...) y que el error se lanza tras el último intento.
Apúntalo a un endpoint que devuelva 401 y confirma que no hay reintentos: falla al primer intento, como debe ser.
Inyecta una fila cuyo campo valor sea "N/A" y comprueba que aparece en quarentena.jsonl mientras el resto se cargó correctamente.
Ejecuta echo $? tras un fallo: debe devolver 1.

Conclusión

Con estas cinco piezas — clasificar el error, reintentar con backoff, aislar las filas malas, cargar dentro de una transacción y fallar de forma visible — tu pipeline ETL deja de necesitar niñera a las tres de la madrugada. El siguiente paso natural es hacerlo idempotente, para que una reejecución tras un fallo no duplique datos. Y un último consejo: cuando te sientas cómodo con el com_retry escrito a mano, prueba la librería tenacity, que hace lo mismo con un decorador. Queda la pregunta: en tu pipeline actual, ¿cuántos de los errores de la última semana eran realmente transitorios?