Documentazione

Tema

Nodo: transform ​

Trasforma uno stream di record in un altro: valida, mappa, espande, filtra, aggrega, ordina. È puro: non tocca sistemi esterni, quindi si prova senza rete. È il nodo più comune di ogni grafo.

0/4
ordini — source ordiniCSV su S3 contratto — transform contrattovalidate a-fatto — transform a-fattomapTransform fatti — sink fattiPostgreSQL scarti — dead-letter scartidead-letter
Scegli una scena e premi ▶ per guardarla scorrere, oppure → per andare avanti un passo alla volta.
Metodo.transform(from, id, transform, { deadLetter?, fromPort?, purpose?, statefulLineage?, writeReport? })
Ingressi1
Uscite≥1 (fan-out libero)
Dead-lettersì, per record (non sulle transform stateful)
CodiceTransform<I, O> in src/core/transform.ts; mapColumns, groupBy, dedup, explode, unpivot, pivot, sort, chunk, parse, serialize, hashRow, surrogateKey, mask, profile, assert in src/transforms/

Quando usarlo ​

Per ogni passaggio che cambia i record usando solo i record stessi e la configurazione. Se il passaggio ha bisogno di leggere da un DB o da un'API è un lookup; se deve scrivere e proseguire col risultato è un'action. Se deve smistare i record su rami diversi è un route, non una serie di filtri.

Un nodo per passaggio: un controllo e una trasformazione sono due nodi, non uno. Il DAG mostrato nell'app è quello eseguito, e deve dire dove un record si è fermato.

Un oggetto per nodo: ogni nodo riceve una transform sua, costruita chiamando la factory nel punto in cui lo si cabla. Lo stesso oggetto su due nodi è un errore di costruzione che nomina i due nodi, perché i due condividerebbero lo stato per-run: tetti di memoria, conteggi, chiavi viste. Due transform distinte con lo stesso nome vanno bene.

Transform pronte ​

FunzioneFa
validate(name, schema)Contratto Zod: il record non conforme lancia ValidationError. Dà le colonne al DAG
mapTransform(name, fn, { outputSchema? })1:1, fn(record, ctx)
mapColumns(name, spec, { rest?, outputSchema? })Select, rename, cast e colonne calcolate in forma dichiarativa (tMap). Dà al DAG le colonne e da quali colonne derivano. Vedi Mappare le colonne
castValue(value, type, { column, empty? })Non è una transform: è il cast stretto di mapColumns da solo, per esempio dentro una derive. Vedi Mappare le colonne
flatMapTransform(name, fn, { outputSchema?, emit? })1:N (tExtractXxxFields): fn(record, ctx) restituisce 0..N output, Iterable o AsyncIterable. Vedi Espansione
explode(name, { field, as?, outputSchema? })Un campo array → un record per elemento, con i campi del padre (tNormalize). Sempre atomico
unpivot(name, { keep, columns, nameAs, valueAs })Largo → lungo (tSplitRow): un record per colonna di columns, col nome in nameAs e il valore in valueAs. Sempre atomico. Vedi Largo e lungo
pivot(name, { key, nameFrom, valueFrom, columns, onConflict?, maxGroups })Lungo → largo (tDenormalize, tPivotToColumnsDelimited): un record per key, una colonna per nome dichiarato. Stateful, maxGroups obbligatorio. Vedi Largo e lungo
filterTransform(name, predicate)Tiene i record per cui predicate(record, ctx) è vero, scarta gli altri
groupBy(name, { key, init, reduce, finalize?, outputSchema? })Aggregazione stateful: un record per gruppo, a stream finito
dedup(name, { key, maxKeys, keep?, duplicates? })Scarta i duplicati per chiave (tUniqRow). Stateful solo con keep: "last"; un'istanza per nodo. maxKeys è obbligatorio: intero ≥ 1, validato alla costruzione
sort(name, { by: [{ key, dir? }], maxRecords })Ordinamento stateful e stabile su più chiavi (tSortRow). maxRecords obbligatorio. Vedi Ordinamento
chunk(name, { size })Record in lotti di size, nell'ordine di arrivo: l'output è readonly I[]. Stateful. Vedi Lotti
parse(name, codec, { bytes, fileName, merge?, emit?, outputSchema? })I byte di un file → una riga per riga del file, con un FileCodec. Espansione atomica di default. Vedi File: parse e serialize
serialize(name, codec, { groupBy?, row?, toOutput, maxRecords, outputSchema? })Record → un file per gruppo (o uno solo), con un FileCodec anche scelto per gruppo. Stateful, maxRecords obbligatorio. Vedi File: parse e serialize
hashRow(name, { columns, as, version?, outputSchema? })Hash di riga delle colonne tracciate in as: v1: + SHA-256, 67 caratteri (tAddCRCRow). Vedi Hash di riga e chiave surrogata
surrogateKey(name, { from, entity, as, outputSchema? })Chiave surrogata deterministica dalla chiave naturale in as: UUIDv5. Vedi Hash di riga e chiave surrogata
mask(name, { columns, key?, onAbsent?, version?, outputSchema? })Maschera le colonne con dati personali: hmac (pseudonimo v1:<keyId>:<hex>), redact, { keep: n }. Vedi Mascheramento e pseudonimizzazione
profile(name, { columns?, distinct, maxColumns? })Profilo delle colonne nel report del nodo: null, min e max, lunghezze delle stringhe, distinti stimati. Pass-through a memoria costante. Vedi Profilo delle colonne
assert(name, { checks, mode?, maxRecords? })Controlli sul dataset a fine stream: conteggi, null, unicità, quadrature. Stateful, gate di default. Vedi Controlli sul dataset
pseudonymize(key, value, version?)Non è una transform: è lo pseudonimo di mask da solo, per cercare a destinazione un valore noto. Vedi Mascheramento e pseudonimizzazione
loadMaskKeyRegistry(env), maskKeyForSpec(id), maskKeyFromBase64(id, valore)Non sono transform: danno la MaskKey di mask, dal registry in run(), segnaposto in getSpec(), o da un segreto che non sta nell'ambiente. Vedi Mascheramento e pseudonimizzazione

Esempi ​

Contratto, mappatura e contratto d'uscita, ognuno nel suo nodo:

ts
.transform("ordini", "ordine-grezzo", validate("ordine-grezzo", ordineGrezzoSchema), {
  deadLetter: scarti, // i record non conformi escono qui, gli altri proseguono
})
.transform("ordine-grezzo", "a-fatto", mapTransform("a-fatto", aFatto, { outputSchema: ordineFattoSchema }))
.transform("a-fatto", "solo-confermati", filterTransform("solo-confermati", (o) => o.stato === "confermato"))
.transform("solo-confermati", "ordine-fatto", validate("ordine-fatto", ordineFattoSchema))

Un'aggregazione:

ts
.transform("righe", "totali-cliente", groupBy("totali-cliente", {
  key: (r: Riga) => r.cliente,
  init: (r) => ({ cliente: r.cliente, totale: 0, righe: 0 }),
  reduce: (acc, r) => ({ ...acc, totale: acc.totale + r.importo, righe: acc.righe + 1 }),
  outputSchema: totaleClienteSchema, // solo per le colonne del DAG
}))

Un'espansione: un ordine diventa una riga per elemento di righe, e il campo array sparisce:

ts
.transform("ordini", "righe", explode("righe", { field: "righe", as: "riga" }), {
  deadLetter: scarti, // `righe` assente, null o non array: ValidationError, il record esce qui
})

Sul ramo catch di un try-catch:

ts
.transform("guardia", "registra-errore", mapTransform<DeadLetter, DeadLetter>("registra-errore", (dl, ctx) => {
  ctx.logger.warn("ordini.errore", { nodeId: dl.error.context.nodeId ?? null });
  return dl;
}), { fromPort: "catch" })

Espansione ​

Un'espansione fa diventare un record d'ingresso zero, uno o più record d'uscita. La domanda è cosa succede se fallisce a metà: gli output già prodotti escono o no? Lo decide emit, per nome e mai dal tipo di ritorno di fn.

emitOutputSe fn lancia un EtlError dopo k output
"atomic" (default)tutti insieme, a espansione finitadi quel record non esce niente; written conta solo le espansioni completate
"streaming"appena prodotti, a memoria costantei k restano a valle (output parziale), written li include, details.emittedBeforeError vale k

In atomico la memoria di un record cresce con la sua espansione: per le espansioni grandi serve emit: "streaming". Per la decodifica di un file c'è parse(codec), un'espansione già scritta (vedi File: parse e serialize). explode è sempre atomico: controlla prima di emettere, e l'array è già in memoria.

ts
.transform("file", "righe", flatMapTransform("righe", (f) => decodifica(f), { emit: "streaming" }), {
  deadLetter: scarti, // obbligatoria sotto un try-catch per-record: vedi Errori
})

explode(name, { field, as? }):

  • array vuoto → nessun output, senza errore;
  • campo assente, null o non array → ValidationError con details: { field, found };
  • l'elemento va in as (default field) e il campo array viene rimosso;
  • as che coincide con un altro campo del padre → ValidationError.

Col grafo in recordIds ogni output ha un id suo e derivedFrom uguale all'id del padre: non serve applyWithLineage.

Mappare le colonne: mapColumns ​

Tenere, rinominare, convertire e calcolare colonne è la parte più grossa di un ETL, e quasi sempre è semplice. mapColumns la esprime come dato invece che come funzione: così il DAG sa quali colonne escono e da quali derivano, senza outputSchema.

ts
.transform("ordini", "mappa", mapColumns("mappa", {
  codice: "id",                                              // rename
  importo: { from: "amount", cast: "number" },               // rename + cast
  consegna: { from: "delivery", cast: "date", empty: "null" },
  totale: {
    derive: (r) => Number(r["qty"]) * Number(r["price"]),
    type: "number",
    from: ["qty", "price"],
  },
}), { deadLetter: scarti })

Senza parametri di tipo il record che derive riceve è Record<string, unknown>: per scrivere r.qty * r.price vedi Tipi più sotto.

mapColumns(name, spec, { rest?, outputSchema? }). Ogni chiave dello spec è una colonna d'uscita:

FormaFa
"colonna"Prende la colonna d'ingresso com'è (rename, o select se il nome è lo stesso)
{ from, cast?, empty? }Prende from e la converte con cast
{ derive: (record) => v, type, from }Calcola il valore. type è il tipo mostrato nel DAG, from le colonne che la funzione legge

derive è una funzione pura del record, come ogni transform. from va dichiarato perché una funzione non si può ispezionare: è il lineage della colonna, e [] vuol dire costante. type è un'etichetta, il valore calcolato non viene controllato.

rest decide delle colonne d'ingresso non dichiarate:

  • "drop" (default): escono solo le colonne dello spec, come una SELECT.
  • "keep": esce tutto il record d'ingresso più le colonne dello spec, che vincono a parità di nome. Un rename in "keep" lascia anche la colonna d'origine, come SELECT *, id AS codice.

Cast. Sono stretti: accettano solo le forme non ambigue, e il resto è un ValidationError.

CastAccettaEsce
numbernumber finito; stringa decimale col punto ("12.5", "-3", "1e3")number
integernumber intero; stringa di sole cifre con segno. Solo interi sicuri (sotto 2^53), e "1.0" non è un interonumber
booleantrue/false, 1/0, "true", "false", "1", "0" (maiuscole ignorate)boolean
stringstringhe com'è; number finito, bigint e boolean con String()string
datestringa YYYY-MM-DD che sia un giorno realestring
timestampDate valida; stringa ISO 8601 con offset esplicito (2026-02-28T10:00:00+02:00)Date
  • In un cast null e undefined passano come null; un rename senza cast lascia il valore com'è. La stringa vuota o di soli spazi è un errore (tranne in string): con empty: "null" diventa null, ed è quello che serve per un CSV.
  • Niente formati localizzati: "1.234,56" e "sì" sono errori. Una sorgente che scrive così si converte in una colonna derive, dove la regola si legge.
  • date è una stringa, non una Date, e una Date in ingresso è rifiutata: il suo giorno dipende dal fuso del processo. Per un istante c'è timestamp, che pretende l'offset.
  • Gli stessi cast si usano da soli con castValue(value, type, { column, empty? }), per esempio dentro una derive.

Errori. Vanno tutti alla dead-letter del nodo, record per record:

  • una colonna d'origine che manca nel record (diversa da una presente con null, che passa): details: { column, output };
  • un cast fallito: details: { column, output, expected, found }. Il messaggio e i details nominano la colonna e i tipi, mai il valore;
  • un EtlError lanciato da una derive: prende nel messaggio la transform e la colonna, e details.output. Qualunque altro errore di derive (un TypeError su un null) è un bug della funzione e ferma la run, come in mapTransform.

Uno spec sbagliato (vuoto, cast sconosciuto, derive senza from o type, empty senza cast, una chiave che la forma non prevede come casst o cast insieme a derive) fa lanciare mapColumns() con un ConfigError già alla costruzione del grafo.

Colonne nel DAG. Con "drop" il nodo dichiara le sue colonne dallo spec, ognuna con le colonne d'origine (sourceColumns): il tipo viene dal cast (integer dà "integer", timestamp dà "date") o da type, ed è unknown per un rename senza cast. Con "keep" il resto del record lo conosce solo chi sta a monte, quindi il nodo eredita le colonne a monte, a meno di un outputSchema: allora le colonne vengono dallo schema, e quelle dello spec portano comunque l'origine. Dettaglio in node-columns.md.

Tipi di mapColumns ​

Senza parametri di tipo l'ingresso è Record<string, unknown> e l'uscita ha le chiavi dello spec, con i tipi dei cast e di derive. Per tipizzare anche l'ingresso i parametri vanno scritti tutti:

ts
const spec = {
  codice: "id",
  importo: { from: "amount", cast: "number" },
  totale: {
    derive: (r) => r.qty * r.price,
    type: "number",
    from: ["qty", "price"],
  },
} as const satisfies MapColumnsSpec<Ordine>;

// Transform<Ordine, { codice: string; importo: number; totale: number }>
mapColumns<Ordine, typeof spec>("mappa", spec);
// con rest: "keep" va scritto anche il terzo parametro
mapColumns<Ordine, typeof spec, "keep">("mappa", spec, { rest: "keep" });

Largo e lungo: unpivot e pivot ​

Il formato largo ha una colonna per misura (mese_01 … mese_12), quello lungo una riga per misura (mese, valore). Excel e i report dei gestionali producono il primo, un database vuole quasi sempre il secondo. unpivot va dal largo al lungo, pivot torna indietro: pivot(unpivot(x)) restituisce x, se la chiave identifica un solo record.

ts
// { reparto, anno, mese_01, mese_02 } → due righe { reparto, anno, mese, valore }
.transform("budget", "per-mese", unpivot("per-mese", {
  keep: ["reparto", "anno"],
  columns: ["mese_01", "mese_02"],
  nameAs: "mese",
  valueAs: "valore",
}), { deadLetter: scarti }) // una colonna assente: ValidationError, il record esce qui

// …e ritorno: una riga per (reparto, anno), una colonna per mese dichiarato
.transform("per-mese", "per-reparto", pivot("per-reparto", {
  key: ["reparto", "anno"],
  nameFrom: "mese",
  valueFrom: "valore",
  columns: ["mese_01", "mese_02"],
  maxGroups: 50_000,
}))

Le colonne d'uscita sono dichiarate in entrambe, mai ricavate dai dati: l'uscita ha esattamente quelle colonne, quindi il DAG le conosce prima della run e non serve un outputSchema. Della transform si sa il nome delle colonne, non il tipo: nel DAG hanno dataType: "unknown", tranne nameAs che è string. Per i tipi, un validate() a valle.

unpivot(name, { keep, columns, nameAs, valueAs }):

  • un record d'uscita per colonna di columns, nell'ordine dichiarato, con le colonne di keep;
  • i campi fuori da keep e columns non passano;
  • una colonna presente con valore null produce la sua riga, con null: ruotare non filtra;
  • una colonna di keep o di columns assente dal record → ValidationError con details: { missing };
  • è record-wise e sempre atomico, come explode: non espone emit, e con recordIds ogni riga deriva dal record padre senza applyWithLineage.

pivot(name, { key, nameFrom, valueFrom, columns, onConflict?, maxGroups }):

  • le righe con la stessa key diventano un record: le colonne di key, poi una colonna per nome di columns col valore di valueFrom. key: [] dà un solo record con tutte le righe;
  • una colonna dichiarata per cui il gruppo non ha righe esce con null;
  • un valore di nameFrom che non è in columns → ValidationError con details: { nameFrom, value }. Non è mai una colonna ignorata in silenzio. Se il valore non è una stringa, details: { nameFrom, found } col tipo trovato;
  • un campo richiesto assente (details: { missing }) o un valore di chiave senza identità confrontabile, cioè oggetto, array, NaN o data non valida (details: { column, found }) → ValidationError. La chiave distingue i tipi: 1 e "1" sono due gruppi;
  • i gruppi escono in ordine di prima comparsa della chiave.

Due righe dello stesso gruppo sulla stessa colonna sono un conflitto, anche se la prima portava null. Decide onConflict:

onConflictEsito
"error" (default)ValidationError con details: { key, column }
"last"resta l'ultimo valore arrivato
(current, incoming)il valore restituito; chiamata solo dal secondo valore di una cella in poi

pivot è stateful: emette a stream finito, quindi un errore di dominio ferma l'intero pivot prima che esca un record, e la deadLetter del nodo non lo intercetta per record (vedi Errori). Con recordIds ogni record d'uscita deriva dalle righe del suo gruppo, comprese quelle sovrascritte da "last"; un valore non dichiarato è attribuito alla riga che lo porta, un conflitto a tutte le righe del gruppo.

Tetto di memoria. maxGroups è obbligatorio e senza default, come sort.maxRecords: pivot trattiene tutti i gruppi fino a fine stream. Passano esattamente maxGroups gruppi. La riga che ne aprirebbe un altro ferma la run senza leggere il resto dello stream, con un Error semplice che nessun guard assorbe. Il tetto conta i gruppi, non le righe: con recordIds ogni gruppo tiene anche gli id delle sue righe.

Una configurazione incoerente fa lanciare unpivot() e pivot() già alla costruzione del grafo: columns vuoto, un nome ripetuto, una colonna sia fra quelle d'identità sia fra quelle ruotate, nameAs o valueAs vuoti, uguali fra loro o a una colonna di keep, nameFrom o valueFrom dentro key o uguali fra loro, maxGroups non intero o minore di 1, un onConflict che non è "error", "last" o una funzione.

Ordinamento ​

sort(name, { by: [{ key, dir? }], maxRecords }) trattiene l'intero stream, lo ordina ed emette. Le chiavi si confrontano in sequenza: la successiva decide solo a parità, ciascuna con la sua dir (default "asc"). Un ordinamento non va riscritto a mano dentro un groupBy o una transform.

ts
// Righe in ordine dentro ogni file; file nell'ordine di arrivo, perché `numeroRiga` riparte da 1 in ognuno.
.transform("esiti", "per-riga", sort("per-riga", {
  by: [{ key: (r: Esito) => r.numeroRiga }],
  maxRecords: 200_000,
}))
.transform("per-riga", "per-file", groupBy("per-file", { key: (r: Esito) => r.fileId, … }))
  • Stabile: a parità di tutte le chiavi resta l'ordine di arrivo. Per questo il sort va prima del groupBy: il groupBy emette i gruppi in ordine di prima comparsa e tiene i record di un gruppo nell'ordine in cui arrivano. I file restano nell'ordine di arrivo solo se partono tutti dallo stesso numero di riga: dopo il sort ogni file compare per la prima volta con la sua riga più bassa, quindi un file che parte da 1 passa davanti a uno che parte da 2, anche se è arrivato dopo.
  • Stringhe per codepoint (l'ordine dei byte UTF-8, come COLLATE "C" di Postgres), senza locale: stesso output su ogni macchina e ogni versione di Node. Quindi "Z" viene prima di "a" ed "é" dopo "z". Non è l'ordine di < di JS, che confronta le unità UTF-16 e mette le emoji prima dei caratteri fra U+E000 e U+FFFF.
  • Date si confronta con getTime().
  • null e undefined sempre in fondo, con asc come con desc: la direzione rovescia i valori, non sposta i buchi. Diverge da Postgres, che con DESC li mette in testa.
  • key viene chiamata una volta per record, all'arrivo, mai dentro il comparatore.
  • Il tipo di una posizione di chiave lo fissa il primo valore non null. Un tipo diverso, NaN o Invalid Date producono un ValidationError sul record che li introduce, con details: { position }. Con recordIds, derivedFrom è l'id di quel record.
  • Lineage 1:1: con recordIds ogni output deriva dal suo input.

Tetto di memoria. maxRecords è obbligatorio e senza default: trattenere è lo scopo del nodo, e quanto pesa un record lo sa solo chi scrive la pipeline. Passano esattamente maxRecords record. All'arrivo del successivo la run si ferma senza leggere il resto dello stream, con un Error semplice che nessun guard assorbe: un ordinamento parziale è peggio di una run fallita. by vuoto, o maxRecords non intero o minore di 1, fanno lanciare sort() già alla costruzione del grafo.

Lotti ​

chunk(name, { size }) raggruppa i record in lotti di size, nell'ordine di arrivo: ogni record d'uscita è un readonly I[]. Serve quando un sink o un'API ha un limite di payload (N record per chiamata, N righe per insert multipla): il lotto diventa un nodo del DAG invece di logica nascosta nel sink.

ts
.transform("ordini", "lotti", chunk<Ordine>("lotti", { size: 500 }))
.sink("lotti", apiBatchSink) // Sink<readonly Ordine[]>: un record = una chiamata

  • Un lotto esce appena è pieno; l'ultimo, se parziale, esce comunque. Stream vuoto → nessun lotto.

  • Stateful: un lotto è fatto di più record, quindi il runner non lo isola per record, nemmeno nello scope di un try-catch per-record o con recordIds. chunk non fallisce mai da sé.

  • Gli errori arrivano a valle, sul lotto: un sink con loadOne che rifiuta un lotto sotto un guard per-record produce un DeadLetter con il lotto intero in raw, e gli altri lotti passano. Con recordIds ogni lotto deriva dagli id dei record che contiene: con guardSnapshots il DeadLetter porta i record d'origine.

  • Per tornare ai singoli record non serve una transform dedicata: una flatMapTransform con l'identità li restituisce nello stesso ordine.

    ts
    .transform("lotti", "record", flatMapTransform<readonly Ordine[], Ordine>("record", (lotto) => lotto))

  • Le colonne del DAG si ereditano da monte: descrivono i record dentro il lotto.

Non è il batchSize di un sink. Con una policy retry il runner spezza in blocchi le chiamate a load di un Sink<T> per ritentarle, ma il sink riceve ancora record singoli. chunk produce record che sono lotti, per un Sink<readonly T[]> in cui un record è una chiamata: il lotto va in dead-letter come un'unità e si vede nel DAG.

Tetto di memoria. size è anche il tetto: il nodo non trattiene mai più di un lotto. size non intero o minore di 1 fa lanciare chunk() con un ConfigError già alla costruzione del grafo. Una finestra temporale (maxWaitMs, lotto parziale dopo un tempo massimo) non c'è: in un runner pull-based richiede timer.

File: parse e serialize ​

I codec del connettore file (csvCodec, csvRawCodec, jsonCodec, xlsxCodec, xmlCodec, parquetCodec) si usano anche in mezzo al grafo, quando i byte non arrivano da fileSource o non vanno a fileSink: un file letto da una cartella Noeva, un allegato di posta, un CSV da consegnare con i suoi metadati. Decodifica e codifica lavorano su byte già in memoria: niente IO, quindi restano transform.

ts
.transform("cartella", "righe", parse<FileNoeva, Record<string, string>, Riga>("righe", csvCodec(), {
  bytes: (f) => f.bytes,           // Uint8Array o AsyncIterable<Uint8Array>
  fileName: (f) => f.name,         // per i details di un errore
  merge: (row, f, n) => ({ ...row, fileId: f.id, riga: n }),
  outputSchema: rigaSchema,        // solo per le colonne del DAG
}), { deadLetter: fileIllegibili }) // il file che non si decodifica esce qui, intero
…
.transform("esiti", "file", serialize<Esito, Consegna, readonly string[]>(
  "file",
  (_key, primo) => csvRawCodec({ delimiter: ";", quote: false, eol: primo.fineRiga }),
  {
    groupBy: (e) => e.fileId,      // senza: un file unico con chiave ""
    row: (e) => rigaDiUscita(e),   // la riga data al codec; default il record
    toOutput: (_key, bytes, esiti, ctx) => consegnaDi(esiti, bytes, ctx.runId),
    maxRecords: 200_000,
  },
))

parse(name, codec, options) è una flatMapTransform sopra codec.decode:

  • emit: "atomic" (default): le righe di un file escono solo a decodifica finita. Un errore a metà file non lascia righe a valle, e con la deadLetter del nodo ci finisce il file intero. emit: "streaming" per i file grandi, con l'output parziale e emittedBeforeError descritti in Espansione, e la stessa regola di validateGraph sotto un try-catch per-record.
  • merge(row, parent, rowNumber) porta nella riga i metadati del file; rowNumber è 1-based e conta le righe decodificate (l'header e le righe vuote saltate dal codec no). Default: la riga com'è.
  • Un errore del codec diventa un ValidationError con details: { fileName, rowNumber, line?, cause }. line è la riga fisica del file, quando il codec la riporta (il CSV sì); cause è il code dell'errore (CSV_INVALID_CLOSING_QUOTE) o il nome della classe (SyntaxError). Il messaggio del codec non viene riportato: JSON.parse e csv-parse citano il contenuto della riga. Un EtlError del codec conserva il tipo e riceve fileName, rowNumber e line. Ogni errore del codec vale come errore del dato, anche un TypeError.
  • merge si chiama per riga: un metadato costoso da ricavare dai byte (un hash del file) va calcolato una volta per file in un mapTransform a monte.
  • Gli errori di merge restano di merge: un EtlError va in dead-letter, qualunque altro ferma la run.
  • Colonne del DAG: da outputSchema; senza, e senza merge, quelle che il codec dichiara (FileCodec.columns, oggi csvCodec con header: false e columns, e xlsxCodec con columns); altrimenti il nodo eredita.

serialize(name, codec, options) trattiene i record ed emette i file a fine stream, perché i gruppi possono intrecciarsi:

  • con groupBy un file per chiave, in ordine di prima comparsa; senza, un file unico con chiave "". Dentro un file i record restano nell'ordine di arrivo. Stream vuoto → nessun file.
  • codec è un FileCodec o una factory (key, primo) => FileCodec, chiamata una volta per file con il suo primo record: serve quando il formato dipende dal dato, come il fine riga del file letto.
  • toOutput(key, bytes, records, ctx) compone il record d'uscita con i byte interi (Uint8Array), i record che compongono il file e il contesto della run.
  • Stateful, con applyWithLineage: con recordIds ogni file deriva dagli id dei record che lo compongono. Un sink che rifiuta un file sotto un guard per-record produce un DeadLetter il cui guardSnapshots porta quei record.
  • Un errore del codec in encode diventa un ValidationError con details: { fileNumber, records, cause }, senza il messaggio del codec; con recordIds è attribuito ai record del file.

Tetto di memoria. maxRecords di serialize è obbligatorio e senza default, contato su tutti i file insieme: all'arrivo del record in più la run si ferma con un Error semplice, mai con file a metà. maxRecords non intero o minore di 1 fa lanciare serialize() con un ConfigError. parse non ha tetto: in atomico trattiene le righe di un file, in streaming nessuna.

Hash di riga e chiave surrogata ​

Due transform record-wise che derivano un valore deterministico dal record e lo aggiungono in as. Il contratto di entrambe è fisso e coperto da vettori di test: i valori finiscono salvati nelle tabelle dei clienti, quindi la regola che decide quando due valori sono uguali (la canonicalizzazione) non cambia mai in silenzio.

ts
.transform("clienti", "chiave", surrogateKey("chiave", {
  from: ["codice", "societa"], // la chiave naturale, nell'ordine che conta
  entity: "cliente",           // la dimensione, non il nome della pipeline
  as: "clienteSk",
}), { deadLetter: scarti })    // un componente null o assente: ValidationError
.transform("chiave", "hash", hashRow("hash", {
  columns: ["ragioneSociale", "citta", "segmento"], // le colonne tracciate
  as: "rowHash",
}), { deadLetter: scarti })

hashRow(name, { columns, as, version?, outputSchema? }) calcola l'hash di riga, per rilevare le righe cambiate quando la sorgente non offre CDC:

  • valore v1: + 64 caratteri esadecimali minuscoli di SHA-256, 67 in tutto; colonna consigliata char(67) in Postgres e in SQL Server;
  • le colonne si codificano per nome: riordinare columns non cambia l'hash, aggiungerne, toglierne o rinominarne una sì (è un cambio di schema);
  • assente, null e stringa vuota danno tre hash diversi; "1" e 1 pure. number intero e bigint invece coincidono;
  • version ha default fisso a 1, anche quando ci sarà una v2: aggiornare la libreria non cambia mai un hash. Non c'è un'opzione algorithm: la versione decide l'algoritmo.

surrogateKey(name, { from, entity, as, outputSchema? }) deriva la chiave surrogata, un UUIDv5 minuscolo:

  • entity (per esempio "cliente") è obbligatoria e non è il nome della pipeline: due pipeline che caricano la stessa dimensione producono le stesse chiavi;
  • i valori di from contano nell'ordine dichiarato e senza nomi: rinominare una colonna non cambia le chiavi, riordinare from sì;
  • colonna uuid in Postgres, uniqueidentifier in SQL Server, senza indice clustered;
  • la codifica è congelata per sempre. Fuori da un nodo (per esempio in un sink) la stessa chiave la dà deriveSurrogateKey(entity, chiaveNaturale), esportata dal barrel.

Entrambe aggiungono la colonna as: senza outputSchema il nodo eredita nel DAG le colonne dell'input, e la colonna nuova non si vede. Conviene dichiararlo quando lo schema d'uscita esiste.

Portare i valori in forma canonica prima è compito della pipeline, perché nessuna delle due transform può farlo al suo posto:

  • date senza ora come stringhe YYYY-MM-DD, non come Date: una Date è un istante, e se arriva dal driver pg dipende dal fuso della macchina. Il connettore Postgres le restituisce già così (date e orari in lettura); da ogni altra sorgente lo fa lo schema;
  • case della chiave naturale normalizzato prima di surrogateKey: con una collation case-insensitive "ABC" e "abc" sono la stessa riga per il MERGE ma due chiavi surrogate;
  • numeric di pg resta stringa, e un number non diventa stringa: i tipi non si fondono.

Errori:

  • un valore non codificabile (Map, Set, istanze di classe, Invalid Date, una stringa con un surrogato spaiato) → ValidationError con la column nei details, sul percorso d'errore del nodo. NaN e ±Infinity invece si codificano, con tag propri;
  • per surrogateKey, un componente null o assente → ValidationError con column, position e found;
  • configurazione validata alla costruzione del grafo, con un Error semplice: columns o from vuoti, duplicati in columns, un nome di colonna o as non in NFC, as vuoto o dentro columns, entity mancante, una version che non esiste.

Mascheramento e pseudonimizzazione ​

mask maschera le colonne con dati personali prima di una destinazione dove il dato in chiaro non serve: analisi, staging, ambienti di demo. È dichiarativa: quali colonne e come sta nella configurazione, quindi si legge e si rivede.

ts
const MASK_KEY = "crm"; // l'id della chiave, costante della pipeline

// in run(): la chiave vera, validata prima che il grafo esista
const key = loadMaskKeyRegistry(process.env).get(MASK_KEY);
// in getSpec(): il segnaposto, senza segreti
const key = maskKeyForSpec(MASK_KEY);

.transform("leads", "maschera", mask("maschera", {
  columns: {
    email: "hmac",       // pseudonimo: v1:crm:<64 hex>
    telefono: "redact",  // "***"
    nome: { keep: 1 },   // "A***"
  },
  key,
}), { deadLetter: scarti })
StrategiaCosa esceNote
"hmac"v1:<keyId>:<64 hex>, varcharPseudonimizzazione deterministica: stesso valore e stessa chiave, stesso pseudonimo. Join e conteggi distinti restano possibili
"redact"***Il valore si scarta, di qualunque tipo: la colonna diventa una stringa
{ keep: n }i primi n caratteri (code point) + ***Solo stringhe; n intero ≥ 1. Una stringa di n caratteri o meno resta leggibile per intero

Il valore fisso *** è la costante MASK_REDACTED. Per tutte e tre: null, undefined e le colonne assenti restano come sono; la stringa vuota è un valore e si maschera.

Attenzione al refuso. Una colonna configurata con un nome che il record non ha (emial, Email al posto di email) non maschera niente e non segnala niente: il dato passa in chiaro. Dove la colonna c'è sempre, onAbsent: "error" trasforma la colonna mancante in un ValidationError; altrimenti metti a monte un validate() con lo schema stretto. Un caso è un errore sempre, anche senza l'opzione: il record ha la colonna ma col nome non in NFC (tipico di un'intestazione CSV scritta su macOS), che non combacerebbe col nome configurato. Le colonne non elencate passano intatte e il record d'ingresso non viene mutato.

Lo pseudonimo (hmac):

  • è HMAC-SHA256 sulla codifica canonica, la stessa di hashRow: 1 e "1" danno pseudonimi diversi, le stringhe si confrontano in NFC senza trim né case folding. Chi vuole il join fra due sorgenti porta prima i valori nella stessa forma (tipo, case, spazi), per esempio con mapColumns;
  • non dipende dal nome della colonna: la stessa email dà lo stesso pseudonimo in ogni colonna e in ogni pipeline che usa la stessa chiave;
  • porta l'id della chiave: ruotarla è aggiungere un id nuovo, e i valori della chiave vecchia restano riconoscibili. Vedi il runbook Chiavi di mask;
  • version ha default fisso a 1, come in hashRow;
  • fuori da un nodo lo stesso valore lo dà pseudonymize(key, valore), esportata dal barrel.

La chiave non è mai una stringa. È una MaskKey opaca: si può usare per firmare, non si può leggere, e in un log, in un JSON o in util.inspect compare solo come [MaskKey crm]. Arriva dal registry loadMaskKeyRegistry(env), che segue la convenzione delle istanze connettore:

  • MASK_KEY_INSTANCES=crm,demo_2027 e una variabile MASK_KEY_<ID>_SECRET per id; senza la lista, MASK_KEY_SECRET vale come chiave default;
  • il valore è base64 di almeno 32 byte (MASK_KEY_MIN_BYTES; openssl rand -base64 32). Un segreto che non sta nell'ambiente (un gestore di segreti, un test) diventa una chiave con maskKeyFromBase64(id, valore), con le stesse regole;
  • una chiave assente, non base64, troppo corta, fatta di byte ripetuti (una passphrase non è una chiave) o uguale a quella di un altro id è ConfigError al caricamento, prima del primo record; il messaggio nomina la variabile, mai il valore.

Dati personali: cosa mask non fa

  • Uno pseudonimo è ancora un dato personale (GDPR art. 4.5). La pseudonimizzazione non è anonimizzazione: chi ha la chiave può ricalcolare lo pseudonimo di un valore noto, e su domini piccoli (un CAP, una data di nascita) invertirlo per enumerazione. La destinazione resta nel perimetro del GDPR.
  • Un DeadLetter contiene il dato in chiaro. raw è il record com'era all'ingresso del nodo che ha fallito: se il DeadLetter nasce a monte di mask, o da mask stesso, i dati personali ci sono tutti. Vale per il sink di dead-letter, per il ramo catch e, con includeRawInReport, per il report persistito. Metti mask prima dei nodi che possono fallire, e proteggi la destinazione degli scarti come la sorgente.
  • Anche senza la chiave uno pseudonimo deterministico dice qualcosa. Su una colonna con pochi valori o molto sbilanciata (un CAP, una data di nascita, un booleano) basta contare le frequenze; chi può far entrare un valore noto nella sorgente ne legge lo pseudonimo a destinazione; e lo stesso valore in due colonne dà lo stesso pseudonimo, quindi le collega. Per le colonne a bassa cardinalità usa redact, non hmac.
  • keep lascia in chiaro una parte del valore, e redact non permette join: nessuna delle due è pseudonimizzazione.

Senza outputSchema il nodo eredita nel DAG le colonne dell'input, coi loro tipi: conviene dichiararlo quando redact cambia il tipo di una colonna. Il DAG non mostra la chiave né il suo id.

Errori:

  • { keep: n } su un valore non stringa → ValidationError con column, reason: "not-a-string" e type nei details (mai il valore), sul percorso d'errore del nodo;
  • una colonna mancante con onAbsent: "error" → ValidationError con reason: "absent-column"; una colonna presente ma non in NFC → ValidationError con reason: "column-not-nfc", sempre;
  • hmac su un valore non codificabile (Map, Set, istanze di classe, Invalid Date, un surrogato spaiato) → ValidationError con la column;
  • configurazione validata alla costruzione del grafo, con ConfigError: columns assente o vuoto, un nome di colonna non in NFC, una strategia sconosciuta, keep non intero o minore di 1, una colonna hmac senza key, una key che non è una MaskKey, una version che non esiste, un onAbsent diverso da "keep" e "error";
  • la chiave segnaposto di maskKeyForSpec usata in una run la ferma prima del primo record: è un errore di cablaggio, non un record da scartare.

Profilo delle colonne ​

profile lascia passare ogni record intatto e ne riassume le colonne nel report del nodo. Serve a vedere nel report quello che i conteggi non dicono: una colonna diventata tutta null, un importo che ha cambiato scala, un codice che ha smesso di variare.

ts
.transform("ordini-validi", "profilo", profile<Ordine>("profilo", {
  columns: ["cliente", "importo", "data"], // assente = tutte le chiavi dei record
  distinct: "approx",                      // obbligatorio: "approx" oppure "off"
}))

Non serve cablare un writeReport: il profilo esce dal report di default che la transform dichiara, e c'è anche quando la run fallisce a valle. In nodeReports["profilo"]:

json
{
  "success": true,
  "profile": {
    "records": 4,
    "distinct": "approx",
    "columns": {
      "cliente": {
        "nulls": 1,
        "number": null,
        "date": null,
        "string": { "count": 3, "minLength": 5, "maxLength": 11 },
        "other": 0,
        "distinctApprox": 2
      },
      "importo": {
        "nulls": 1,
        "number": { "count": 3, "min": -3, "max": 200 },
        "date": null,
        "string": null,
        "other": 0,
        "distinctApprox": 3
      },
      "data": {
        "nulls": 1,
        "number": null,
        "date": {
          "count": 3,
          "min": "2025-12-31T00:00:00.000Z",
          "max": "2026-03-01T00:00:00.000Z"
        },
        "string": null,
        "other": 0,
        "distinctApprox": 3
      }
    }
  }
}

Per ogni colonna:

CampoContenuto
nullsValori null o undefined. Un campo assente dal record conta come null
numberConteggio, min e max dei numeri finiti; null se la colonna non ne ha
dateConteggio, min e max (ISO 8601) delle Date valide. Una data arrivata come stringa è una stringa
stringConteggio, minLength e maxLength in unità UTF-16: le lunghezze, mai i valori
otherTutto il resto: booleani, bigint, NaN, infiniti, date non valide, oggetti, array
distinctApproxDistinti stimati fra i valori scalari non null; null con distinct: "off". Oggetti e array non entrano nella stima

Da sapere:

  • Memoria costante: profile non trattiene record. Tiene pochi numeri per colonna e, con distinct: "approx", uno sketch HyperLogLog da 4 KiB per colonna. Non è stateful: emette in streaming e resta isolabile per record.
  • Distinti approssimati: l'errore standard è circa l'1,6 % e la stima sta entro il ± 5 % dal valore vero; sulle cardinalità piccole è praticamente esatta. L'hash non è crittografico: è una misura di osservabilità, non un controllo di unicità. Per quello c'è dedup.
  • Senza columns il nodo profila tutte le chiavi proprie dei record, scoprendole man mano: una colonna comparsa tardi conta come null nei record già passati. Le colonne scoperte hanno un tetto di memoria, maxColumns (default 256): superato, la run si ferma con un Error semplice che nomina l'opzione, e nessun try-catch lo assorbe. Le colonne dichiarate non hanno tetto.
  • Non rifiuta mai un record: non ha una dead-letter da riempire, e un record che non è un oggetto conta con ogni colonna null.
  • Le date sono istanze Date: una data arrivata come stringa (per esempio date e timestamp senza fuso dal connettore Postgres) è profilata come stringa, quindi per lunghezza.
  • Il report non porta name: in nodeReports la chiave è già l'id del nodo.
  • Un'istanza per nodo: lo stato è per-run e per-istanza. Lo stesso oggetto profile su due nodi sommerebbe i due flussi in un profilo solo: la costruzione lo rifiuta.
  • Un writeReport passato al nodo sostituisce il profilo, non lo unisce.

Cosa finisce nel report

Il report viene salvato e mostrato nell'app. Il valore di una stringa non ci entra mai. Gli estremi di numeri e date invece sono valori dei dati: un importo massimo, una data minima, senza il record a cui appartengono. Se anche quelli sono sensibili, dichiara columns e lascia fuori la colonna. Con columns assente finiscono nel report anche i nomi delle chiavi dei record. I valori più frequenti non ci sono: esporrebbero dati reali.

Controlli sul dataset: assert ​

validate() giudica un record alla volta. assert giudica il dataset: fa passare i record intatti e, a fine stream, confronta ogni check con la sua soglia. Ferma i caricamenti sbagliati nell'insieme: file arrivato vuoto o troncato, quota di null anomala, chiave non univoca, totali che non quadrano.

ts
.transform("clienti-validi", "qualita", assert<Cliente>("qualita", {
  checks: [
    rowCount({ min: 1 }),
    nullRatio("email", { max: 0.05, severity: "warn" }),
    unique("id"),
    custom<Cliente, number>("quadratura importi", {
      init: () => 0,
      step: (totale, c) => totale + c.importo,
      verdict: (totale) => ({ observed: totale, ok: totale === totaleAtteso }),
      threshold: { expected: totaleAtteso },
    }),
  ],
}))

I check ​

CheckOsservaSupera se
rowCount({ min?, max? })il numero di recordsta fra min e max; ne serve almeno uno
nullRatio(colonna, { max })la quota di null della colonna, fra 0 e 1non supera max
unique(colonna | colonne[], { maxKeys? })i record la cui chiave era già comparsasono zero
custom(nome, { init, step, verdict, threshold? })ciò che restituisce verdictverdict restituisce ok: true

Ogni check accetta severity: "error" (default) ferma la run, "warn" la lascia proseguire.

  • Soglie inclusive: max: 0.05 accetta esattamente il 5 %.
  • Null è null, undefined o chiave assente. La stringa vuota è un valore.
  • nullRatio su zero record supera, con osservato null: che zero record siano un problema lo dice rowCount.
  • unique ignora i record con la chiave null; con una chiave composta basta una colonna null. Riporta quanti duplicati ha visto, mai quali.
  • unique ricorda le chiavi viste, in entrambe le modalità: ha un tetto di memoria proprio, maxKeys (default 100 000). Superarlo ferma la run anche se il check è warn: è un tetto di memoria, non un esito.
  • custom è un reducer puro: niente IO, niente stato fuori da quello che init crea. L'osservato è un numero o un booleano, perché finisce nel report. threshold serve solo al report e al messaggio d'errore.
  • I nomi dei check sono derivati (rowCount, nullRatio(email), unique(anno,numero)); custom vuole un nome esplicito. Due check con lo stesso nome nello stesso assert fanno lanciare assert() alla costruzione.

Gate o streaming ​

mode decide quando escono i record.

  • "gate" (default) trattiene tutti i record finché il verdetto non è noto, e li emette nello stesso ordine solo se nessun check error è fallito. Se fallisce, a valle non arriva niente. Costa memoria: fino a maxRecords record (default 100 000). Oltre il tetto la run si ferma con un Error semplice, che nessun try-catch assorbe, prima che a valle arrivi un record.
  • "streaming" emette i record man mano e lancia dopo l'ultimo. Non trattiene niente, ma quando il verdetto arriva il caricamento a valle è già avvenuto.

La regola: streaming solo se il caricamento a valle è annullabile. Questa è la matrice, e vive solo qui:

Caricamento a valle dell'assertStreaming protegge?Perché
Sink Postgres, SQL Server o MySQL senza retrySìl'errore annulla la transazione unica
Sink su uno staging seguito da un post-task di scambioSìil post-task non parte: i dati rifiutati restano nello staging, la tabella in uso è intatta. Anche con retry
Sink con una policy di retry o rateLimit, senza scambioNoi blocchi già scritti sono confermati
arraySink, sink di file e di cartellaNole righe sono già scritte, e oggi non esiste un task di scambio per i file
Nodo action fra l'assert e il sinkNogli effetti sono già avvenuti (mail spedita, chiamata fatta): uno scambio non li annulla
Qualunque sink dentro lo scope di un try-catchNon si costruisce, salvo lo scambiovalidateGraph rifiuta il grafo: il guard assorbirebbe il verdetto. Lo ammette solo con lo scambio, vedi sotto

In gate nessuna riga è un problema: il sink non riceve niente finché il verdetto non è positivo.

Streaming con lo scambio (write-then-swap). Il sink scrive su uno staging, e un post-task lo scambia con la tabella in uso solo se il sink è riuscito. Un verdetto negativo lo impedisce in entrambi i casi:

  • fuori da un guard la run fallisce, e il post-task non parte;
  • dentro un guard il verdetto viene catturato e il sink scrive tutto, ma il suo input risulta troncato: il post-task è saltato e la run è partial. validateGraph ammette questa forma solo se ogni sink a valle dell'assert ha un edge di controllo verso un task, e a valle non c'è nessuna action.
ts
.tryCatch("ordini", "guard", { mode: "circuit-breaker" })
.transform("guard", "qualita", assert("qualita", { mode: "streaming", checks: [unique("id")] }))
.sink("qualita", staging) // l'id del nodo è staging.name: "carica-staging"
.sink("guard", errori, { fromPort: "catch" })
.task("scambia", scambia, { after: ["carica-staging"] })

⚠️ validateGraph controlla la forma, non l'intenzione: non sa se il task fa davvero uno scambio, né se il sink scrive su uno staging e non sulla tabella in uso. Un ANALYZE come post-task soddisfa la regola e non protegge niente. E lo scambio protegge la tabella in uso, non lo staging: dopo un verdetto negativo lo staging contiene i dati rifiutati, fino al pre-task della run successiva. Come scriverlo: runbook.

Un gate con soli check warn non può fermare niente, quindi non trattiene: lascia passare i record man mano, senza tetto. Il report mostra la modalità effettiva, "streaming".

Esito ​

CasoEsito
Tutti i check superatii record proseguono; run success
Un check warn fallitoi record proseguono; l'esito va nel report e in ctx.logger.warn. Lo stato della run non cambia
Un check error fallito, fuori da un guardAssertionError; run failed
Un check error fallito, gate dentro un guardun solo DeadLetter sul catch, con raw: undefined; zero record a valle; run partial
Superato maxRecords o maxKeysError semplice: run failed, anche dentro un guard

L'AssertionError elenca tutti i check error falliti, ciascuno con osservato e soglia. Non è un ValidationError: quello riguarda un record contro il suo schema, questo il dataset.

Dentro un try-catch, il sink a valle di un gate fallito è saltato: non chiama load e nel report ha skipped: { because: "<id dell'assert>" }. Senza questa regola un sink di file riscriverebbe il file vuoto. Vedi sink. Un post-task di quel sink è saltato a sua volta.

Nel report ​

Non serve cablare un writeReport: l'esito esce dal report di default della transform, anche quando la run fallisce. In nodeReports["qualita"]:

json
{
  "success": false,
  "assert": {
    "mode": "gate",
    "records": 1200,
    "verdict": "error",
    "checks": [
      {
        "check": "rowCount",
        "kind": "rowCount",
        "severity": "error",
        "observed": 1200,
        "threshold": { "min": 1 },
        "outcome": "ok"
      },
      {
        "check": "nullRatio(email)",
        "kind": "nullRatio",
        "column": "email",
        "severity": "warn",
        "observed": 0.08,
        "threshold": { "max": 0.05 },
        "outcome": "warn"
      },
      {
        "check": "unique(id)",
        "kind": "unique",
        "column": "id",
        "severity": "error",
        "observed": 3,
        "threshold": { "max": 0 },
        "outcome": "error"
      }
    ]
  }
}

verdict è il peggiore degli esiti. Se lo stream non è arrivato in fondo (un tetto superato, un errore a monte) il verdetto non esiste: verdict e ogni outcome valgono "unknown", con observed: null.

Il report non contiene dati di record: solo nomi di check e di colonne, numeri e booleani.

Da sapere:

  • È stateful: deve vedere l'intero stream, e il runner non la isola per record. Una deadLetter sul nodo resta vuota.
  • Lineage 1:1: con recordIds ogni record in uscita deriva dal suo record d'ingresso, quindi un errore a valle dell'assert risale al record giusto.
  • Un'istanza per nodo: lo stato è per-run e per-istanza.
  • Un writeReport passato al nodo sostituisce l'esito dei check, non lo unisce.
  • Nel DAG un nodo in gate mostra i record estratti che crescono e zero scritti fino alla fine: è il comportamento atteso, non un blocco.

Una transform su misura ​

ts
const primoPerCliente: Transform<Ordine, Ordine> = {
  name: "primo-per-cliente",
  stateful: true, // accumula stato sull'intero stream
  async *apply(records) {
    const visti = new Set<string>();
    for await (const r of records) {
      if (!visti.has(r.cliente)) {
        visti.add(r.cliente);
        yield r;
      }
    }
  },
};

Per il caso generale — togliere i duplicati per chiave — c'è già dedup, con la scelta dell'occorrenza da tenere e il tetto sulle chiavi in memoria: non riscriverlo.

  • stateful: true quando la transform deve vedere tutti i record insieme (aggregazioni, deduplica, ordinamento) o quando un suo output combina più record d'ingresso (lotti): il runner non la esegue mai un record alla volta, nemmeno sotto un try-catch.
  • columns (per esempio outputSchemaColumns(schema)) quando cambia la forma del record.
  • emit: "streaming" quando la transform può far uscire dei record e poi fallire sul record che li ha prodotti (un output parziale): validateGraph la tratta come un'espansione in streaming. emit: "before-verdict" quando fa uscire i record e li giudica tutti insieme a fine stream, come assert in streaming: validateGraph la rifiuta dentro qualunque try-catch, a meno che ogni sink a valle sia seguito da un post-task di scambio.
  • applyWithLineage per dichiarare quali record d'ingresso hanno prodotto ogni output: serve a correlare un errore a valle di un'aggregazione al gruppo colpevole (groupBy, pivot, chunk, serialize e assert lo fanno).
  • held per dichiarare in lavorazione, quando la transform trattiene record per costruzione: quanti ne ha ricevuti e non ha ancora fatto uscire, letto dal runner a ogni cadenza dei contatori dal vivo (sort, groupBy, pivot, chunk, serialize e assert in gate lo fanno, come le transform che il builder genera per lookup, ai e action). Si dichiara solo se si sa davvero: senza, il nodo non ha il campo, che non vuol dire zero. Vedi sotto, «In lavorazione di una transform».
  • writeReport per dare al nodo un report di default con le metriche della transform, senza che chi cabla il grafo debba ricordarsene (profile, assert e il lookup lo fanno). Lo stato da cui attinge vive in una WeakMap chiavata sul RunContext, non in una closure. Il runner lo chiama anche quando il nodo fallisce.

Metriche proprie di una transform ​

Una transform che vuole riportare qualcosa oltre ai quattro contatori del runner (extracted, written, rejected, rejectedReasons) segue una convenzione sola:

  1. lo stato sta in una WeakMap<RunContext, …> creata dalla factory della transform, e apply lo aggiorna. Se la transform ha anche applyWithLineage, lo aggiorna anche quello: con recordIds il runner usa il secondo, e un conteggio sul solo apply resterebbe a zero in silenzio. Vale anche per chi avvolge una transform di libreria (serialize, groupBy);
  2. la transform dichiara un writeReport che lo legge, e restituisce i valori di default quando la run non ha mai invocato apply;
  3. chi cabla il nodo non passa un writeReport proprio, oppure chiama quello della transform dentro il suo.
ts
function contaPerTipo(name: string): Transform<Evento, Evento> {
  const perRun = new WeakMap<RunContext, Readonly<Record<string, number>>>();
  return {
    name,
    async *apply(records, ctx) {
      for await (const evento of records) {
        const totali = perRun.get(ctx) ?? {};
        perRun.set(ctx, {
          ...totali,
          [evento.tipo]: (totali[evento.tipo] ?? 0) + 1,
        });
        yield evento;
      }
    },
    writeReport: (ctx) => ({ perTipo: { ...(perRun.get(ctx) ?? {}) } }),
  };
}

Una variabile chiusa nella funzione che costruisce il grafo, mutata da una transform e letta dal writeReport di un altro nodo, non è un'alternativa: niente vieta di eseguire due volte lo stesso grafo, e in quel caso la seconda run partirebbe dai totali della prima. Lo stato è per-run e per-istanza: una chiamata alla factory per nodo.

In lavorazione di una transform ​

Una transform che trattiene record — li riceve e li fa uscire dopo, tutti insieme o a gruppi — lo dichiara con held(ctx), e la UI mostra quanti ne sta tenendo. Il valore è per run, come le metriche: heldCounter() del core tiene la WeakMap sul RunContext, apply chiama set quando trattiene o rilascia, e read va in held.

ts
function ultimoPerCliente(name: string): Transform<Ordine, Ordine> {
  const held = heldCounter();
  return {
    name,
    stateful: true,
    held: held.read,
    async *apply(records, ctx) {
      const ultimi = new Map<string, Ordine>();
      try {
        for await (const r of records) {
          ultimi.set(r.cliente, r);
          held.set(ctx, ultimi.size);
        }
        let restanti = ultimi.size;
        for (const r of ultimi.values()) {
          restanti -= 1;
          held.set(ctx, restanti);
          yield r;
        }
      } finally {
        held.set(ctx, 0); // anche se la run si ferma a metà
      }
    },
  };
}
  • Si abbassa prima di emettere: il record che esce non è più trattenuto qui.
  • Torna a zero in un finally, così uno stream interrotto non lascia un numero appeso.
  • Se c'è applyWithLineage, aggiorna il contatore anche lì: con recordIds il runner usa quello.
  • Non si dichiara su una transform record per record (una map, un filtro, un'espansione): "entrati − usciti" comprenderebbe i record scartati o moltiplicati, e il numero mentirebbe.

Errori ​

CasoEsito
EtlError (es. ValidationError) con deadLetteril record esce come DeadLetter con raw, lo stream continua
EtlError sotto un try-catch, senza deadLetteril record va sul ramo catch
EtlError senza nessuno dei duefail-fast: la run si ferma
errore non EtlError (un TypeError)la run si ferma sempre: è un bug, non un dato
espansione atomica che fallisce dopo k outputdi quel record non esce niente; il padre va in dead-letter
espansione in streaming che fallisce dopo k outputi k restano a valle; il padre va in dead-letter con details.emittedBeforeError: k
espansione in streaming senza deadLetter (o stateful) nello scope di un try-catch per-recordvalidateGraph rifiuta il grafo: un record passerebbe sia dal try sia dal catch
assert in streaming con un check error nello scope di un try-catch, qualunque sia il suo modevalidateGraph rifiuta il grafo: il guard assorbirebbe il verdetto
lo stesso oggetto transform su due nodivalidateGraph rifiuta il grafo e nomina i due nodi: condividerebbero lo stato per-run

Una transform stateful (groupBy, pivot, sort, chunk, serialize, assert, dedup con keep: "last") non si isola mai per record: spezzarla ne falserebbe il risultato. La sua deadLetter, anche se configurata, resta quindi vuota, e un suo errore prende una delle altre due strade:

  • dentro lo scope di un try-catch va sul ramo catch, catturato in stile circuit-breaker: il DeadLetter ha raw: undefined e, se l'errore è emerso mentre il nodo leggeva ancora l'ingresso, lastConsumed con l'ultimo record letto;
  • altrimenti è fail-fast: la run si ferma.

dedup con keep: "first" (il default) invece non è stateful: tiene le chiavi per run, resta isolabile, e con duplicates: "error" ogni duplicato arriva nella dead-letter del nodo col payload.

Opzioni del nodo ​

OpzioneDefaultEffetto
deadLetter—Sink dei record che falliscono in questo nodo; resta vuoto su una transform stateful
fromPort—Porta del nodo a monte ("catch" per il ramo errori di un try-catch, una porta di un route)
purpose—Lo scopo del nodo per chi usa la pipeline: vedi Scopo del nodo
statefulLineage"last"Su una transform stateful senza lineage proprio: "last", "all" o "none" (con recordIds)
writeReport—Riassunto del nodo nel report. Sostituisce quello dichiarato dalla transform (profile, assert, lookup), non lo unisce

Nel DAG e nel report ​

Le colonne del nodo vengono da validate() o da outputSchema (per parse anche dalle colonne dichiarate dal codec); senza, il nodo eredita quelle a monte, anche se la forma del record è cambiata. Nel report, rejected e rejectedReasons contano gli scarti in dead-letter.

Nei contatori dal vivo, sort, groupBy, pivot, chunk, serialize e assert in gate mostrano anche in lavorazione: i record trattenuti e non ancora usciti. Lo dichiarano anche le transform che il builder genera per lookup, ai e action. Le altre transform non lo hanno.

In nodeReports il nodo porta, nell'ordine: il writeReport passato al nodo, altrimenti quello dichiarato dalla transform (profile, assert, lookup), altrimenti { name, success }. Vedi run-report.

Anti-pattern ​

  • Un fetch, una query o una scrittura dentro una transform: è IO nascosto, invisibile al DAG e a una review. Va in un lookup, un'action o un sink.
  • Un try/catch JavaScript che scarta in silenzio: ogni scarto deve essere contato e ispezionabile.
  • Validate e map nello stesso nodo.

Noeva è un marchio registrato di 4D S.R.L.

© 2026 4D S.R.L. · Gruppo4D