Como fazer logging num pipeline ETL: passo a passo

Um pipeline ETL que corre em silêncio é um problema à espera de acontecer: quando algo falha a meio da noite, não sabes onde nem porquê. Adicionar logging a um pipeline ETL dá-te um registo claro de cada fase — extração, transformação e carga — para detetares erros cedo, medires a duração e saberes quantas linhas foram processadas. Ao contrário de um simples print(), o logging dá-te níveis de gravidade, data e hora automáticas e a possibilidade de escrever em vários destinos ao mesmo tempo. Vais construir, passo a passo, um sistema simples de logging e monitorização em Python, usando apenas a biblioteca-padrão.

Pré-requisitos

• Python 3.9 ou superior instalado.
• Um pipeline ETL, mesmo que pequeno (ler dados, transformar, gravar).
• Conhecimentos básicos de funções em Python.
• Um editor de código, como o VS Code.

Passo 1: Configurar o logging básico

O módulo logging faz parte da biblioteca-padrão do Python, por isso não precisas de instalar nada. Define primeiro o formato das mensagens e o nível mínimo que queres registar. O nível INFO é ideal para acompanhar o progresso no dia a dia; guarda o DEBUG para quando estás mesmo a investigar um 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")

Passo 2: Registar cada fase do pipeline

Regista o início de cada fase e, sobretudo, o número de linhas processadas. Estas contagens são a forma mais simples de monitorização: se um dia extraíres 0 linhas onde costumavas ter milhares, o log denuncia logo o problema. Usa o nível INFO para o fluxo normal e o nível WARNING quando algo é estranho mas não impede o pipeline de continuar.

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

Passo 3: Guardar os logs em ficheiro com rotação

Ver as mensagens no ecrã ajuda durante o desenvolvimento, mas em produção precisas de as guardar. O RotatingFileHandler escreve num ficheiro e cria automaticamente um novo quando o atual atinge o tamanho-limite, evitando ficheiros gigantes que enchem o 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)

Passo 4: Capturar e registar erros

Envolve cada fase num bloco try/except. O método logger.exception() regista a mensagem e o traceback completo, o que torna a análise do erro muito mais rápida. Relança a exceção a seguir para o pipeline parar de forma controlada, em vez de continuar com dados incompletos. Assim, os erros nunca passam despercebidos e ficam guardados com todo o 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

Passo 5: Medir a duração de cada execução

A duração é uma métrica de monitorização muito útil: um pipeline que costuma demorar 2 minutos e passa a demorar 30 é um sinal de alerta claro. Usa time.perf_counter() para medir e registar o tempo total de cada execução.

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

Verificar o resultado

Corre o pipeline e abre o ficheiro etl.log. Deves ver uma linha por cada fase, com data, hora, nível e mensagem — incluindo as contagens de linhas e a duração final. Para testar o tratamento de erros, força uma falha de propósito (por exemplo, aponta a leitura para um ficheiro que não existe) e confirma que aparece uma linha ERROR com o traceback. Depois, filtra só os problemas com um comando rápido:

grep ERROR etl.log

Conclusão

Com poucas linhas de código passaste de um pipeline "caixa preta" para um processo observável: sabes o que corre, quanto demora e onde falha. O passo seguinte natural é enviar estes logs para uma ferramenta central, como o Azure Monitor ou o Application Insights, e criar alertas automáticos sempre que surgir um ERROR. Qual das métricas achas mais importante vigiar no teu pipeline: o número de linhas processadas ou o tempo de execução?