ETL: extrair, transformar, carregar

O formato de um pipeline que roda todo dia — idempotência, falha parcial, reprocessamento e linhagem.

Um ETL é um programa que roda de novo, todo dia, sobre dados que mudam. Isso é tudo o que o distingue de um script de análise, e é o que cria os problemas dele: o que acontece quando roda duas vezes? E quando falha no meio?

As três etapas, separadas de propósito#

adopt Arcane.IO as IO

// EXTRAIR — só lê. Nenhuma regra de negócio aqui.
action extrair(caminho):
    yield IO.read_csv(caminho, yes)

// TRANSFORMAR — só calcula. Nada de arquivo, nada de banco.
action transformar(linhas):
    limpas := [l cycle l in linhas given (l["valor"] ?? "") isnt ""]
    yield [l with {"valor": float(l["valor"])} cycle l in limpas]

// CARREGAR — só escreve.
action carregar(linhas, destino):
    IO.write_csv(destino, linhas)
    yield len(linhas)

A separação não é estética. transformar não toca em disco, e por isso ela é a única das três que se testa sem preparar ambiente — e é onde mora toda a regra que pode estar errada.

Rodar duas vezes tem de dar o mesmo resultado#

É a propriedade que mais falta num ETL: idempotência. Sem ela, reprocessar um dia que falhou duplica tudo o que já tinha entrado.

adopt Arcane.Database as DB

action carregar(db, linhas, dia):
    DB.transacao(db, lambda:
        [DB.upsert(db, "vendas", l, ["id"]) cycle l in linhas])
    yield len(linhas)

A transação é o que torna "carregou tudo" e "não carregou nada" as duas únicas saídas possíveis. Sem ela, uma falha no meio deixa metade dentro — e a segunda execução não tem como saber qual metade.

Falhar bem#

Um ETL que quebra às três da manhã precisa dizer onde parou e o que já tinha feito:

adopt Arcane.Logging as Log

action rodar(caminho, destino):
    Log.info($"etl: comecando {caminho}")
    monitor:
        brutas := extrair(caminho)
        Log.info($"etl: {len(brutas)} linhas lidas")

        limpas := transformar(brutas)
        Log.info($"etl: {len(limpas)} linhas apos limpeza")

        gravadas := carregar(limpas, destino)
        Log.info($"etl: {gravadas} gravadas em {destino}")
        yield gravadas
    handle Error as e:
        Log.erro($"etl falhou em {caminho}: {e.message}")
        trigger $"o ETL de {caminho} nao terminou"

O trigger de fora preserva o erro original em .causa — o relatório mostra as duas camadas, e quem lê o log de manhã vê tanto o que o pipeline tentava fazer quanto o que o impediu.

A linha que não entra#

Descartar silenciosamente uma linha ruim é a forma mais comum de um relatório ficar errado sem ninguém notar. Separe, conte e guarde:

action separar(linhas):
    boas := []
    ruins := []
    cycle l in linhas:
        given (l["valor"] ?? "") is "":
            ruins.append(l with {"motivo": "valor vazio"})
        otherwise:
            boas.append(l)
    yield {"boas": boas, "ruins": ruins}

resultado := separar(brutas)
given len(resultado["ruins"]) > 0:
    IO.write_csv("rejeitadas.csv", resultado["ruins"])
    out $"atencao: {len(resultado['ruins'])} linha(s) rejeitadas"

O arquivo de rejeitadas é o que transforma "os números não batem" numa investigação de cinco minutos.

Ordem, dependência e paralelismo#

Quando um pipeline tem etapas independentes, elas não precisam esperar umas às outras. Mas o ganho depende de onde está o custo:

adopt Arcane.Concurrent as P

// as três fontes são independentes: baixam juntas
parallel:
    vendas := baixar("vendas")
    clientes := baixar("clientes")
    produtos := baixar("produtos")

// a transformação é CPU: processos de verdade
limpas := P.map_processos(transformar_lote, em_lotes(vendas, 10000))

Agendar#

O pipeline em si não deveria saber a que horas roda. Quem agenda é de fora — cron, um orquestrador, ou Arcane.Concurrent.repetir_a_cada num processo que fica de pé:

dataforge run etl/diario.df --data=2026-09-16

Receber a data como argumento é o que torna o reprocessamento possível: rodar o dia 3 de novo é trocar um número, e não mexer no código.

A lista de conferência de um ETL#

• Rodar duas vezes dá o mesmo resultado?
• A carga está dentro de uma transação?
• As linhas rejeitadas são contadas e gravadas?
• O log diz quantas linhas entraram em cada etapa?
• A data é argumento, e não hoje() no meio do código?
• A transformação é testável sem disco e sem banco?

Por onde seguir#