Tipi di nodo

Ogni processo è un grafo costruito col DSL graph()…build(), e ogni passaggio del grafo è un nodo di uno di questi undici tipi, più il nodo pipeline, che istanzia una sub-pipeline e alla costruzione diventa i nodi che contiene. Questa pagina è il cheatsheet: tutto quello che serve per scegliere il nodo giusto e scriverne la firma. Ogni tipo ha poi una scheda di dettaglio, con opzioni, errori ed esempi, che si apre con una breve demo animata del nodo al lavoro.

Codice in src/core/ (graph.ts per i tipi, graph-builder.ts per il DSL, graph-builder-options.ts per le sue opzioni). Una pipeline importa tutto dalla radice del pacchetto, @noeva/etl-core; dentro il core lo si raggiunge dal barrel src/core/index.ts.

Indice

Nodo	Metodo	Cosa fa	Ingressi	Uscite	Dead-letter
source	.source(id, source)	Estrae da un sistema esterno, in streaming	0	≥1	no
transform	.transform(from, id, t, opts)	Trasforma record → record, puro (niente IO)	1	≥1	sì
lookup	.lookup(from, id, config, opts)	Legge in sola lettura da un DB o un'API, a batch di chiavi	1	≥1	sì
ai	.ai(from, id, config, opts)	Chiede a un modello, la risposta arricchisce il record	1	≥1	sì
action	.action(from, id, action, opts)	Scrive per record e prosegue col risultato	1	≥1	sì
route	.route(from, id, { routes }, opts)	Manda ogni record su una sola porta con nome	1	≥1 per porta, default compresa	sì
join	.join(id, { build, probe, … })	Accoppia due flussi per chiave (inner/left)	2 (build + probe)	≥1	sì (lato probe)
union	.union(id, { from, order })	Mette in fila N flussi con la stessa forma, record invariati	≥2	≥1	no
try-catch	.tryCatch(from, id, opts)	Apre uno scope di gestione errori sui nodi a valle	1	≥1 su out + esattamente 1 su catch	no
sink	.sink(from, sink, opts)	Carica su un sistema esterno, in modo idempotente	1	0	no
task	.task(id, task, opts)	Esegue un'operazione senza record (truncate, scambio, operazioni su fogli e colonne Excel)	0 (solo after)	0	no
pipeline	.pipeline(from, id, sub, opts)	Istanzia una sub-pipeline, appiattita alla costruzione	1	≥1 per uscita della sub-pipeline	sì, solo root

Quale nodo mi serve?

Devo…	Nodo
leggere dati da un DB, un file, una casella di posta, un workspace Noeva	source
validare, mappare, filtrare, aggregare	transform
tenere, rinominare, convertire o calcolare colonne	transform con mapColumns
espandere un record in N (righe d'ordine, elementi di un array)	transform
passare da colonne a righe e viceversa (mese_01…mese_12 ↔ mese/valore)	transform con unpivot e pivot
ordinare i record (un file per un cliente, un report)	transform con sort
mandare i record a un sink o a un'API in lotti di N	transform con chunk
decodificare i byte di un file arrivati da un'altra source (CSV, JSON, XLSX…)	transform con parse
comporre file dai record (uno per gruppo) per un sink che vuole i byte	transform con serialize
rilevare le righe cambiate senza CDC (hash delle colonne tracciate)	transform con hashRow
dare a una dimensione chiavi surrogate stabili fra le run	transform con surrogateKey
mascherare o pseudonimizzare colonne con dati personali prima di un sink	transform con mask
vedere nel report null, estremi e distinti di ogni colonna	transform con profile
vedere nel report che cosa un nodo ha fatto a ogni record	recordReport su lookup, ai, action
fermare un caricamento se il dataset è vuoto, pieno di null o con duplicati	transform con assert
arricchire ogni record con dati di un DB o di un'API (grande, selettivo)	lookup
ricavare un dato che nessuna query sa dare: estrarre, classificare	ai
arricchire con un secondo flusso che sta in memoria	join
scrivere a metà grafo e usare il risultato (un id, un esito)	action
fare più scritture in ordine sullo stesso record	più action in catena
mandare ogni record su un ramo diverso secondo una condizione	route
mandare lo stesso record a più rami	nessun nodo: fan-out, più nodi con lo stesso from
riunire N flussi con lo stesso tracciato (N workspace, N file)	union
non fermare la run per un record che fallisce in un solo nodo	deadLetter sul nodo
gestire gli errori di un tratto di grafo, e decidere dove vanno i record	try-catch
scrivere il risultato finale	sink
conservare la storia di una dimensione (SCD tipo 2)	sink con mode: "scd2"
svuotare uno staging prima di caricare, scambiare le tabelle dopo	task con after
caricare un ramo prima di un altro (le dimensioni prima dei fatti)	after sulla source del secondo ramo: task
riusare un pezzo di grafo, con un contratto d'ingresso e d'uscita	pipeline, che istanzia una sub-pipeline

Scelte che si confondono spesso:

• lookup o join? Il join tiene in memoria tutto il lato build: va bene se è piccolo. Il lookup interroga solo le chiavi che servono: è la scelta quando il lato di destra è grande, o quando un record porta più chiavi candidate.
• action o sink? Se la scrittura è l'ultima cosa che succede al record, è un sink. Se a valle serve il suo risultato, è un'action. Se è una lettura, è un lookup.
• route o fan-out? Il fan-out manda lo stesso record a tutti i rami; il route lo manda a uno solo. Più filterTransform al posto di un route valutano il predicato N volte, non garantiscono rami esclusivi e perdono i record che non passano nessun filtro.
• union o join? Il join accoppia due record per chiave; il union mette in fila i record così come sono.
• task, action o sink? Un task non vede record: gira una volta per run, quando i nodi in after sono riusciti. Se l'operazione riguarda ogni record è un action o un sink.

Firme del builder

Ogni metodo restituisce un nuovo builder: si concatena. build() valida la forma e lancia GraphValidationError al primo problema.

graph(name, { recordIds?, commitOn?, runParams? })   // runParams: parametri di lancio
  .source(id, source, { after?, purpose?, writeReport? })
  .transform(from, id, transform, { deadLetter?, fromPort?, purpose?, statefulLineage?, writeReport? })
  .lookup(from, id,
    { resolver, key, merge, onNotFound?, batchSize?, cache?, maxCacheEntries?, outputSchema?, retry?, rateLimit? },
    { deadLetter?, fromPort?, purpose?, recordReport?, writeReport? })
  .ai(from, id, config, { deadLetter?, fromPort?, purpose?, recordReport?, writeReport? })   // stessa config del lookup
  .action(from, id, action, { deadLetter?, fromPort?, outputSchema?, purpose?, recordReport?, writeReport? })
  .route(from, id, { routes: { porta: (r, ctx) => boolean } }, { deadLetter?, fromPort?, purpose?, writeReport? })
  .join(id, { build, probe, how, buildKey, probeKey, merge, onDuplicateBuildKey?, bufferProbe? },
    { deadLetter?, purpose?, writeReport? })
  .union(id, { from: [id | { node, port }, …], order? }, { purpose?, writeReport? })
  .tryCatch(from, id, { mode?, includeRawInReport?, guardSnapshots?, snapshotLimit?, onSnapshotOverflow?, purpose?, writeReport? })
  .sink(from, sink, { fromPort?, purpose?, retry?, rateLimit?, writeReport? })
  .task(id, task, { after?, purpose?, retry?, rateLimit?, writeReport? })
  .pipeline(from, id, sub, { params?, deps?, fromPort?, deadLetter?, purpose? }) // sub = subPipeline({ name, description, input, outputs, params?, deps?, build })
  .build(); // → GraphPipeline: .run(ctx) e .getSpec()

⚠️ .sink() non prende un id: l'id del nodo è sink.name. Due sink con lo stesso name collidono con «Id nodo duplicato»: dai nomi distinti ("ordini", "ordini_scartati").

Opzioni comuni:

Opzione	Dove	Effetto
deadLetter	transform, lookup, ai, action, route, join; pipeline solo nella root, distribuita ai nodi isolabili dell'istanza	Il record che fallisce va su questo sink, lo stream continua
fromPort	chi parte da un route, da un try-catch o da un'istanza	Da quale porta d'uscita del nodo a monte prendere i record
purpose	tutti; su pipeline è lo scopo dell'istanza	Lo scopo del nodo per chi usa la pipeline, nel popover della UI: vedi Scopo del nodo
writeReport	tutti	Riassunto del nodo nel report, a fine nodo (mai il dato grezzo). Sostituisce quello dichiarato dalla transform
retry	lookup, ai, sink e task con retrySafe	Ritenta gli errori transitori con backoff esponenziale
rateLimit	lookup, ai, sink e task con retrySafe	Chiamate al secondo, al più
after	source, task	Edge di controllo: il nodo parte solo dopo che i nodi elencati (task o sink) sono riusciti
outputSchema	lookup, ai, action, mapTransform, mapColumns, flatMapTransform, explode, groupBy, parse, serialize, hashRow, surrogateKey, mask	Colonne del nodo nel DAG, quando cambia la forma del record

Scopo del nodo

Con purpose un nodo dice cosa fa e perché sta in questa pipeline, a chi la usa e non a chi l'ha scritta. Nell'app, il pannello di dettaglio del nodo mostra un'icona di info accanto al titolo: cliccata, apre un popover con lo scopo. Senza scopo l'icona non c'è, ma in una pipeline del catalogo un nodo senza scopo fa fallire etl catalog check (regola scopo-dei-nodi), che elenca tutti i mancanti insieme (esenti i nodi che genera il core: il collettore e i confini di un'istanza). Nel sito delle docs lo stesso testo compare nella tabella «Lo scopo dei nodi», sotto il grafo di ogni scheda pipeline e sub-pipeline.

.transform("righe", "solo-fatturati", filterTransform(...), {
  purpose:
    "Tiene solo gli ordini già fatturati, gli unici che il gestionale sa riconciliare.",
})

• Italiano, testo semplice (niente markdown né HTML: la UI lo rende come testo), da una a tre frasi. In termini di business: niente SQL, nomi di variabili d'ambiente, né ciò che la UI mostra già (colonne, connettore).
• Appartiene al nodo nella sua pipeline, non al componente: lo stesso filtro ha uno scopo diverso in ogni pipeline. Non ha default.
• build() rifiuta uno scopo vuoto dopo il trim o più lungo di 400 caratteri (PURPOSE_MAX_LENGTH), con un errore che nomina il nodo, con l'id qualificato se sta dentro un'istanza.
• In una sub-pipeline, i nodi del corpo lo dichiarano nella definizione e l'istanza al punto di chiamata; la description della sub-pipeline non fa da default. Dettaglio nella scheda pipeline.
• Il collettore dei dead-letter, il nodo dead-letter che il piano aggiunge per ogni sink di scarto, non si crea col builder: lo scopo glielo dà il core, uguale in ogni pipeline (DEAD_LETTER_PURPOSE, in src/core/purpose.ts).
• Sta nel piano, sul nodo, solo quando c'è. Non entra nello snapshot del piano, e il gate delle versioni non lo conta come cambio di comportamento: dare o riscrivere uno scopo non chiede un bump. Perché il gate lo riconosca, lo scopo è una stringa letterale fra le opzioni del metodo; se le opzioni del nodo sono calcolate, stendile: { ...opzioni, purpose: "…" } vale opzioni (versioni di pipeline).

Porte

• Porte d'uscita (fromPort): le dichiarano solo try-catch (out, predefinita, e catch) e route (una per condizione, più default, sempre esplicita). Un nodo senza porte rifiuta fromPort; una porta che non esiste è un errore che elenca quelle valide. Un nodo pipeline ha come porte le uscite della sub-pipeline: con una sola uscita fromPort si omette.
• Porte d'ingresso: solo il join ne ha (build e probe), e il builder le mette da sé.
• Fan-out: più nodi con lo stesso from (e la stessa porta) ricevono tutti ogni record. Il ramo più veloce può andare avanti di 16 record sul più lento; i record in attesa sono «in lavorazione» sul nodo a monte, nei contatori dal vivo.
• Fan-in: solo join e union. Gli altri nodi hanno esattamente un ingresso, tranne source e task, che non hanno ingressi di dati.

Edge di controllo

Oltre agli edge di dati, lungo cui scorrono i record, esistono gli edge di controllo: non portano record, dicono solo «B parte dopo che A è riuscito». Si scrivono con after su .source() e .task(), partono da un task o da un sink, e arrivano a un task o a una source. Un nodo che non parte perché il suo predecessore non è riuscito è saltato, con tutti i suoi discendenti di dati. Dettagli nella scheda task.

Errori: chi li gestisce

Solo gli EtlError (ValidationError, LookupError, ActionError, JoinError, …) si possono gestire; un bug (TypeError) interrompe sempre la run. Tre meccanismi, dal più stretto al più largo:

Meccanismo	Come	Effetto
Dead-letter del nodo	{ deadLetter: sink } sul nodo	Il record fallito esce come DeadLetter, lo stream continua
Scope try-catch	.tryCatch(…) a monte	Gli errori dei nodi a valle arrivano sul ramo catch
Fail-fast	nessuno dei due (il default)	La run si ferma con status: "failed"

Nel piano ogni sink di scarto diventa un nodo dead-letter, il collettore, con lo scopo fisso del core (vedi Scopo del nodo) e un edge da ogni nodo che lo alimenta.

Il fail-fast vale per tutto il grafo: al primo errore fatale il runner annulla anche i rami indipendenti, aspetta che ogni sink e ogni task chiuda, e solo allora riporta la run fallita. Un task che fallisce è sempre fatale: non sta mai nello scope di un try-catch.

Uno scope try-catch cattura solo gli errori nati nei nodi a valle della sua porta out: un errore che arriva da monte risale e ferma la run, anche se attraversa lo scope. Un nodo con dead-letter propria resta nello scope, ma i suoi errori per record finiscono prima nella sua dead-letter. source e sink non hanno dead-letter: un sink con loadOne diventa isolabile per record solo sotto un try-catch.

Le regole che build() fa rispettare

• almeno un source e almeno un sink; un source non ha ingressi, un sink non ha uscite;
• ogni nodo è raggiungibile da un source e alimenta qualcosa a valle, tranne il task, che non ha edge di dati;
• un edge di controllo parte da un task o da un sink e arriva a un task o a una source, senza cicli con gli edge di dati e senza il deadlock sul fan-in descritto nella scheda task;
• un solo ingresso per nodo, tranne join (uno build e uno probe) e union (≥2);
• ogni porta di un route è collegata, default compresa; il try-catch ha esattamente un edge su catch e almeno uno su out;
• un join con build e probe che discendono dallo stesso nodo richiede bufferProbe: true; un union in concat con ingressi da un antenato comune è rifiutato;
• retry e rateLimit su un sink o un task che non dichiara retrySafe sono rifiutati;
• una source incrementale ha una dichiarazione watermark valida (colonna, tipo, initial del tipo giusto, lookback solo sui timestamp), e commitOn: "partial" è ammesso solo se il grafo ha almeno una source incrementale;
• uno scopo (purpose) non è vuoto dopo il trim e non supera i 400 caratteri;
• niente cicli, niente id duplicati o sconosciuti.

Un grafo minimo

import {
  graph,
  validate,
  mapTransform,
  type DeadLetter,
  type Sink,
  type Source,
} from "@noeva/etl-core";

export function buildOrdiniPipeline(
  sources: { ordini: Source<unknown> },
  sinks: { fatti: Sink<OrdineFatto>; scarti: Sink<DeadLetter> },
) {
  return graph("ordini-giornalieri")
    .source("ordini", sources.ordini)
    .transform(
      "ordini",
      "contratto-grezzo",
      validate("contratto-grezzo", ordineGrezzoSchema),
      {
        deadLetter: sinks.scarti,
      },
    )
    .transform("contratto-grezzo", "a-fatto", mapTransform("a-fatto", aFatto))
    .transform(
      "a-fatto",
      "contratto-fatto",
      validate("contratto-fatto", ordineFattoSchema),
    )
    .sink("contratto-fatto", sinks.fatti)
    .build();
}

Un nodo per passaggio: il DAG mostrato nell'app è quello eseguito, e deve dire dove un record si è fermato. Per vederli tutti insieme in un grafo interattivo: Il grafo, nodo per nodo.