Técnicas

Streams

Processamento incremental de dados grandes com generators encadeados.

O problema#

Um arquivo de log com um milhão de linhas. Você quer os erros. A forma ansiosa:

dataforge

linhas := IO.read("app.log").lines()        # 1 milhão de strings
registros := linhas >> morph interpretar     # mais 1 milhão de vaults
erros := registros >> sift e: e["nivel"] is "ERROR"

Três cópias completas dos dados na memória — e talvez você só queira ver o primeiro erro.

A forma incremental#

dataforge

stream action linhas_do_log():
    cycle l in IO.read("app.log").lines():
        emit l

stream action interpretar(fonte):
    cycle linha in fonte:
        partes := linha.split(" ")
        emit {"data": partes[0], "nivel": partes[1],
              "mensagem": partes.slice(2).join(" ")}

stream action apenas(fonte, nivel):
    cycle registro in fonte:
        given registro["nivel"] is nivel:
            emit registro

erros := apenas(interpretar(linhas_do_log()), "ERROR")

Montar a cadeia não lê nada. Uma linha entra, atravessa os três estágios, sai — e só então a próxima começa. A memória usada é a de uma linha.

O ganho em first()#

dataforge

primeiro := apenas(interpretar(linhas_do_log()), "ERROR").first()

Isso lê até o primeiro erro e para. Se ele estiver na linha 3, as outras 999.997 nunca são tocadas.

O padrão de três estágios#

Estágio	Faz	Não faz
Origem	produz os itens brutos	não interpreta
Transformação	dá estrutura a cada item	não valida
Filtro	descarta o que não interessa	não decide o destino

Separados assim, cada estágio é testável e reutilizável isoladamente. Quando o formato de entrada mudar de CSV para JSON, só o primeiro muda.

Erros que atravessam o pipeline#

O ponto mais delicado de um ETL: uma linha ruim não pode derrubar as outras. A solução é fazer o erro viajar como um dado:

dataforge

stream action separar(fonte):
    cycle linha in fonte:
        campos := linha.split(",")
        given len(campos) is 5:
            emit campos
        otherwise:
            emit {"__erro__": $"campos de menos: '{linha.trim()}'"}

Cada estágio seguinte reconhece o marcador e o repassa intacto:

dataforge

match item:
    point {"__erro__": e}:
        emit item              # passa adiante sem tocar
    point [nome, email, idade, setor, salario]:
        emit {...}             # processa normalmente

No fim, o consumidor separa os dois fluxos. Ninguém perde dado e ninguém para o processamento por causa de uma linha torta. Um monitor em volta de tudo abortaria o pipeline inteiro na primeira falha.

Agregar sem materializar#

dataforge

contagem := {}
cycle registro in interpretar(linhas_do_log()):
    nivel := registro["nivel"]
    contagem[nivel] := contagem.get(nivel, 0) + 1

O acumulador cresce com o número de níveis distintos (3), não com o número de linhas. Essa é a diferença entre um agregado e uma cópia.

Normalizar na entrada#

dataforge

"nome": nome.trim().title(),        # "  ANA SILVA " → "Ana Silva"
"email": email.trim().lower(),      # " ana@X.COM " → "ana@x.com"

Dados de fora chegam sujos. Limpar uma vez, na fronteira, evita ter que lembrar disso em cada consulta depois.

Converter só depois de validar#

dataforge

given not registro["idade"].isdigit():
    problemas.append("idade nao numerica")
...
"idade": cast registro["idade"] as Integer

A ordem importa: cast "abc" as Integer dispara erro. Verificar primeiro transforma uma exceção num registro rejeitado com mensagem clara.