ETL (Extract, Transform, Load) é o padrão que move dados de fontes brutas para destinos analíticos. Com Python, você pode construir um pipeline ETL completo sem ferramentas pagas: extrai de arquivos, APIs ou bancos, transforma com Pandas e carrega onde precisar — tudo com logging, tratamento de erros e agendamento.

Estrutura do pipeline

# etl_vendas.py
import pandas as pd
import logging
import os
from datetime import datetime
from sqlalchemy import create_engine

logging.basicConfig(
    filename=f"log_etl_{datetime.now():%Y%m%d}.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def extrair():
    ...

def transformar(df):
    ...

def carregar(df, engine):
    ...

def main():
    ...

if __name__ == "__main__":
    main()

E — Extrair

def extrair():
    logging.info("Iniciando extração")
    dfs = []

    # De arquivos Excel em uma pasta:
    pasta = "dados_brutos"
    for arq in os.listdir(pasta):
        if arq.endswith(".xlsx"):
            df = pd.read_excel(os.path.join(pasta, arq))
            df["arquivo_origem"] = arq
            dfs.append(df)
            logging.info(f"  Lido: {arq} ({len(df)} linhas)")

    df_bruto = pd.concat(dfs, ignore_index=True)
    logging.info(f"Extração concluída: {len(df_bruto)} linhas totais")
    return df_bruto

T — Transformar

def transformar(df):
    logging.info("Iniciando transformação")
    n_inicial = len(df)

    # Limpeza:
    df = df.drop_duplicates(subset=["id_pedido"])
    df = df.dropna(subset=["valor", "cliente"])
    df.columns = [c.lower().strip().replace(" ", "_") for c in df.columns]

    # Tipos:
    df["valor"] = pd.to_numeric(df["valor"], errors="coerce")
    df["data_pedido"] = pd.to_datetime(df["data_pedido"], dayfirst=True)

    # Colunas derivadas:
    df["ano_mes"] = df["data_pedido"].dt.to_period("M").astype(str)
    df["comissao"] = df["valor"] * 0.05
    df["categoria"] = pd.cut(
        df["valor"],
        bins=[0, 500, 2000, 10000, float("inf")],
        labels=["Bronze", "Prata", "Ouro", "Diamante"]
    )

    n_final = len(df)
    logging.info(f"Transformação: {n_inicial} -> {n_final} linhas (-{n_inicial-n_final} removidas)")
    return df

L — Carregar

def carregar(df, engine):
    logging.info("Iniciando carga")

    # Banco de dados:
    df.to_sql("vendas_processadas", engine, if_exists="replace",
              index=False, chunksize=10000, method="multi")

    # Excel consolidado:
    df.to_excel("relatorio_consolidado.xlsx", index=False)

    # Resumo para outra tabela:
    resumo = df.groupby(["ano_mes", "categoria"])["valor"].agg(
        total="sum", qtd="count"
    ).reset_index()
    resumo.to_sql("resumo_mensal", engine, if_exists="replace", index=False)

    logging.info(f"Carga concluída: {len(df)} registros gravados")

Main: orquestrando e tratando erros

def main():
    inicio = datetime.now()
    logging.info(f"Pipeline ETL iniciado: {inicio}")

    try:
        engine = create_engine(os.environ["DATABASE_URL"])
        df_bruto = extrair()
        df_clean = transformar(df_bruto)
        carregar(df_clean, engine)

        duracao = (datetime.now() - inicio).seconds
        logging.info(f"Pipeline concluído com sucesso em {duracao}s")

    except Exception as e:
        logging.exception(f"Pipeline falhou: {e}")
        raise

Agendando com cron (Linux/Mac)

# Adicionar ao crontab (crontab -e):
# Executar todo dia às 06:00:
# 0 6 * * * /usr/bin/python3 /caminho/etl_vendas.py

# No Windows, use o Agendador de Tarefas ou Task Scheduler via PowerShell

Perguntas frequentes

Quando usar Python vs ferramentas de ETL visuais (Airbyte, Fivetran)?

Para integrações com conectores prontos e alto volume, ferramentas dedicadas são mais eficientes. Python ETL customizado é ideal quando a lógica de transformação é complexa e específica do negócio, quando as fontes são não convencionais (arquivos internos, sistemas legados sem API) ou quando o custo das ferramentas não se justifica.

Idempotência: como garantir que rodar duas vezes não duplique dados?

Use if_exists="replace" ao gravar tabelas derivadas, ou implemente chave única na tabela de destino e use INSERT ... ON CONFLICT DO NOTHING via SQLAlchemy. Também registre o processamento por data para reprocessar apenas períodos alterados.