parse, sort e serialize. Da estratti conto in CSV, con i movimenti di più conti mescolati, un CSV per conto con i movimenti in ordine di data.| Nodo | Tipo | Scopo |
|---|---|---|
estratti | source | Legge gli estratti conto di prova della dimostrazione, ognuno un file CSV con i movimenti di più conti. |
righe | transform | Apre ogni estratto e ne ricava i movimenti, uno per riga, ricordando da quale file e da quale riga viene ciascuno. Un file che non si legge finisce per intero fra gli scarti. |
movimenti | transform | Controlla che ogni movimento abbia conto, data e importo validi. Un movimento fuori regola finisce fra gli scarti, e gli altri proseguono. |
ordina | transform | Mette i movimenti in ordine di conto e, per ogni conto, di data, così ogni file per conto li elenca dal più vecchio. |
componi-file | transform | Scrive un file CSV per conto, con i suoi movimenti in ordine e gli importi con due decimali. |
file-per-conto | sink | Consegna i file per conto, uno per ciascun conto che compare negli estratti. |
scarti | dead-letter | Raccoglie i record scartati dai nodi di questa pipeline, ognuno con il motivo dello scarto. |
Il grafo è una catena; lo scopo di ogni nodo è nella tabella sotto il grafo. Qui il dettaglio tecnico:
righe — parse con csvCodec({ delimiter: ";" }): i byte arrivano dal record (bytes), non da fileSource, ed è il caso di una cartella Noeva o di un allegato di posta. La decodifica non fa IO, quindi resta una transform pura. merge porta nella riga il nome del file e il numero di riga. Con emit: "atomic", il default, le righe di un file escono solo a decodifica finita: un errore porta il file intero nella dead-letter, con fileName, rowNumber e line nei details, e non lascia righe a metà.movimenti — validate con movimentoSchema, con la dead-letter scarti.ordina — sort su due chiavi, conto e data, stabile. È stateful: trattiene tutto lo stream fino alla fine, con il tetto maxRecords obbligatorio (100 000), e non ha dead-letter.componi-file — serialize con groupBy per conto: un file per gruppo, a fine stream, nell'ordine di prima comparsa della chiave. Dopo il sort i file escono in ordine di conto. row porta l'importo a stringa con due decimali, toOutput compone il record del file.estratti: due estratti conto di prova, in memoria, ognuno con l'header conto;data;descrizione;importo e i movimenti di due conti. Uno dei movimenti ha un importo che non è un numero.file-per-conto: un record per conto, con il nome del file (movimenti-<conto>.csv), il numero di movimenti e i byte del CSV. In una pipeline vera qui c'è un sink di file, per esempio una cartella Noeva.scarti: la dead-letter di righe e movimenti, con l'envelope DeadLetter di ogni file o riga scartati.I byte dipendono solo dagli estratti: lo stesso ingresso dà gli stessi file, nello stesso ordine. Un sink di file che sovrascrive per nome, o che salta un upload con gli stessi byte come noevaFileSink in new-version, rende sicura la riesecuzione.
movimentoSchema: file, riga, conto, data, descrizione, importo (con coercizione da stringa, perché il CSV non ha tipi).filePerContoSchema: le colonne del record in uscita da componi-file, per il DAG.| Caso | Dove finisce | Esito della run |
|---|---|---|
| Un file che non si decodifica | scarti, per intero, con fileName e rowNumber | partial |
| Una riga fuori contratto | scarti, la riga sola | partial |
Più movimenti di maxRecords | la run si ferma: niente ordinamento a metà | failed |
Senza un sink di scarto, i primi due casi fermano la run.
Nessuno: il dataset è in memoria.
pipeline.test.ts copre i file per conto in ordine di data, una riga scartata, un file che non si decodifica e la riesecuzione con gli stessi byte.
Prima versione: non c'è una versione precedente da confrontare.