Como tratar erros e retries num pipeline ETL em Python

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

Um pipeline ETL que corre bem no portátil parte em produção por razões banais: a API responde 429, a base de dados fecha a ligação a meio, um ficheiro traz uma linha corrompida. Tratar erros e retries num pipeline ETL é o que separa um script frágil de um processo em que se pode confiar. O objetivo aqui é claro: distinguir uma falha temporária de um erro real, repetir só quando faz sentido e nunca perder dados em silêncio.

Pré-requisitos

Python 3.9 ou superior instalado.
As bibliotecas requests e sqlalchemy (pip install requests sqlalchemy).
Um pipeline ETL simples já a funcionar (extrair de uma API, transformar, carregar numa base de dados).
Noções básicas de try/except em Python.

Passo 1: Separar erros transitórios de erros permanentes

Nem todos os erros merecem uma nova tentativa. Um timeout ou um HTTP 503 são transitórios: daqui a dez segundos o pedido provavelmente funciona. Já um HTTP 401 (credenciais erradas) ou um KeyError numa transformação são permanentes — repetir mil vezes dá sempre o mesmo erro e só atrasa o pipeline.

Começa por escrever essa regra em código, para deixar de adivinhar:

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

Passo 2: Criar uma função de retry com backoff exponencial

Repetir de imediato, cinco vezes seguidas, é a melhor forma de irritar uma API que já está sobrecarregada. A boa prática é o backoff exponencial: esperar 1s, 2s, 4s, 8s... e juntar um pouco de aleatoriedade (jitter) para vários processos não repetirem todos ao mesmo 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)

Repara em dois detalhes importantes: se o erro não for transitório, o raise acontece logo na primeira tentativa; e na última tentativa o erro também sobe, em vez de ficar escondido.

Passo 3: Aplicar o retry à extração

A extração é onde o pipeline toca no mundo exterior, por isso é aí que os retries valem mais. Envolve o pedido HTTP numa função sem argumentos e passa-a ao 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 sempre o timeout. Sem ele, um pedido pendurado bloqueia o pipeline durante horas e nenhum retry chega a acontecer. Se a API devolver o cabeçalho Retry-After, respeita esse valor em vez do backoff calculado.

Passo 4: Isolar as linhas más em quarentena

Na transformação o problema é outro: uma única linha com o campo valor a dizer "N/A" não deve deitar abaixo um lote de 200 000 registos. Em vez de rebentar, separa o que é válido do que não é e guarda os rejeitados numa dead-letter queue (um ficheiro ou uma tabela de quarentena) com o motivo do erro.

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")

Uma regra saudável: se a quarentena passar de uma percentagem que definas (por exemplo 5% do lote), o pipeline deve falhar. Muitas linhas más de repente não é um acidente, é um contrato de dados que mudou.

Passo 5: Falhar de forma controlada e visível

A carga deve ser transacional: ou entra tudo, ou não entra nada. E quando o pipeline falha mesmo, tem de falhar alto — com código de saída diferente de zero, para o orquestrador (Airflow, Azure Data Factory, cron) perceber que algo correu 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 no fim, rollback se houver erro
        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 o resultado

Não esperes pela produção para descobrir se o tratamento de erros funciona. Testa-o de propósito:

Aponta o URL para um endpoint que devolve 503 e confirma nos logs as esperas a crescer (1s, 2s, 4s...) e o erro final ao fim das tentativas.
Aponta para um endpoint que devolve 401 e confirma que não há repetições — falha à primeira, como deve ser.
Injeta uma linha com o campo valor a "N/A" e verifica que aparece no ficheiro quarentena.jsonl e que as restantes foram carregadas.
Corre echo $? depois de uma falha: deve devolver 1.

Conclusão

Com estas cinco peças — classificar o erro, repetir com backoff, isolar linhas más, carregar em transação e falhar de forma visível — o teu pipeline ETL deixa de precisar de babysitting às três da manhã. O passo seguinte natural é tornar o pipeline idempotente, para que uma re-execução depois de uma falha não duplique dados. E uma dica final: quando estiveres confortável com o com_retry escrito à mão, experimenta a biblioteca tenacity, que faz o mesmo com um decorador. Fica a pergunta: no teu pipeline atual, quantos dos erros da última semana eram mesmo transitórios?