Técnicas

Pipelines e orquestração

DAG, dependências, retry e carga incremental — sem servidor, sem agendador, sem banco de metadados.

Airflow, Prefect e Dagster resolvem orquestração com um servidor, um banco de metadados e um agendador. Arcane.Pipeline resolve com uma estrutura de dados e um laço — o que cabe num processo só, que é onde a maioria dos pipelines de verdade vive.

dataforge

adopt Arcane.Pipeline as P

fluxo := P.fluxo("vendas")

P.etapa(fluxo, "extrair", extrair)
P.etapa(fluxo, "limpar",   limpar,   ["extrair"])
P.etapa(fluxo, "conferir", conferir, ["limpar"])
P.etapa(fluxo, "carregar", carregar, ["conferir"])

relatorio := P.rodar(fluxo)

A ordem sai das dependências, não da ordem em que você declarou — é a ordenação topológica que todo DAG faz. P.grafico(fluxo) mostra o resultado antes de rodar.

As três garantias#

1. Um ciclo é erro, não aviso#

text

P.etapa(fluxo, "x", acao_x, ["y"])
P.etapa(fluxo, "y", acao_y, ["x"])

erro: as etapas x, y dependem umas das outras.
  nota: um ciclo não tem ordem possível
  dica: quebre o ciclo, ou junte as etapas numa só

Não há ordem que satisfaça as duas. Escolher uma arbitrariamente produziria um resultado que ninguém consegue explicar — e que muda entre execuções.

2. Falhou? quem depende é PULADO#

text

ok      extrair
falhou  limpar     banco fora do ar
pulada  conferir   depende de limpar
pulada  carregar   depende de conferir
ok      notificar

3. Retry para a falha passageira#

dataforge

// 3 tentativas, esperando 2s, 4s entre elas
P.etapa(fluxo, "carregar", carregar, ["limpar"], 3, 2)

A espera cresce a cada tentativa. Se o banco está ocupado, insistir no mesmo ritmo mantém ele ocupado. E o relatório diz em qual tentativa passou — uma etapa que sempre precisa de três é um problema que a média esconde.

Carga incremental#

Reprocessar tudo a cada execução é o que faz um pipeline de 10 minutos virar um de 6 horas em dois anos.

dataforge

fluxo := P.fluxo("vendas", "estado/vendas.json")

action extrair(ctx):
    desde := P.marca(fluxo, "ate") ?? "1970-01-01"
    novas := Banco.consultar(db,
        "SELECT * FROM vendas WHERE atualizado_em > ?", [desde])
    given len(novas) bigger 0:
        P.marcar(fluxo, "ate", maior(novas, "atualizado_em"))
    yield novas

Rodar só um pedaço#

dataforge

P.rodar_ate(fluxo, "limpar")   // roda 'extrair' e 'limpar', e para

Ele resolve as dependências sozinho — roda o que limpar precisa, e nada além. É como se depura um pipeline longo sem esperar a carga.

O relatório#

json

{
  "fluxo": "vendas",
  "ok": yes,
  "duracao": 12.4,
  "resumo": {"total": 4, "ok": 4, "falhou": 0, "pulada": 0, "saltada": 0},
  "etapas": [
    {"etapa": "extrair", "estado": "ok", "duracao": 8.1, "tentativas": 1},
    …
  ],
  "resultados": {"extrair": [ … ], "limpar": [ … ]}
}

O relatório é o produto. Sem ele, saber o que aconteceu exige ler log — e log de pipeline é o que ninguém lê até quebrar.

Condição: o mesmo fluxo em modos diferentes#

dataforge

P.etapa(fluxo, "carga_completa", completa, ["limpar"],
        1, 0, lambda ctx: MODO is "cheio")

A etapa é saltada quando a condição dá falso — e saltar não é falhar: o fluxo segue verde. É como se roda o mesmo pipeline em modo cheio e incremental sem duplicá-lo.