Pipeline ETL: impianto e strumenti di verifica

Deliverable D2.2.1: verifica e validazione tecnica delle pipeline ETL

Questo capitolo documenta l'impianto delle pipeline ETL consegnato per il Data Lake MAPS e gli strumenti con cui le pipeline vengono verificate e accettate. L'impianto comprende il modello a flussi, il pattern Medallion con lo schema EAV, le utility condivise e un'implementazione di riferimento; gli strumenti di verifica comprendono un motore di verifica automatica, una skill operativa e una dashboard di tracciamento. La piattaforma traccia 232 dataset su 27 fonti, di cui 100 prioritari (dashboard ETL, sincronizzazione del 12 giugno 2026); il codice attualmente versionato conta circa 38 unità-pipeline su 17 fonti (gst-maps-pipelines v0.38.4). La differenza tra i due conteggi è di granularità: la dashboard traccia i dataset all'unità di lavorazione, incluse le fonti non ancora codificate, mentre il codice conta le directory-pipeline, e le famiglie che adottano un orchestratore condiviso coprono molti dataset con poche unità. I dati sono normalizzati a granularità comunale o sovracomunale e archiviati nel layer Silver EAV. Lo sviluppo per-dataset dei flussi è a cura del team di sviluppo; la guida operativa completa — convenzioni, pattern, procedure di deployment e un'implementazione di riferimento — è disponibile nell'Appendice A2 — Manuale operativo per lo sviluppo di pipeline ETL.

Le cinque fasi della pipeline

Ogni dataset percorre cinque fasi. Le prime quattro sono flussi ETL standardizzati che separano l'acquisizione dei dati dalla trasformazione, dalla validazione e dalla catalogazione; la quinta integra il dataset nel Datastore, rendendolo fruibile. La separazione tra i flussi è deliberata: se la logica di trasformazione cambia, è sufficiente rieseguire il solo flow di trasformazione senza riscaricare i dati sorgente; se una regola di validazione viene aggiornata, viene avviato il solo flow di qualità. Questa indipendenza rende lo sviluppo iterativo più rapido e riduce il rischio di reintrodurre errori nelle fasi non modificate.

I quattro flussi ETL sono sviluppati dal team di sviluppo, che segue le linee guida e le pipeline già realizzate come riferimento:

1. Flow 1 — Ingestion: si identifica la fonte dati, si implementa la logica di download o di scraping e si scrivono i file grezzi nel Bronze layer. Il flow si considera completo quando almeno un file ben formato è disponibile al percorso Bronze atteso e registrato in bronze.ingestion_log.
2. Flow 2 — Transform: si leggono i file Bronze, si risolvono gli identificatori territoriali tramite territory_resolver, si mappano le colonne sorgente sugli attributi EAV e si scrive in silver.territory_attributes. È il flow più specifico per ogni dataset.
3. Flow 3 — Data quality: un framework di validazione custom verifica sia i file Bronze sia i record Silver rispetto ai criteri di accettazione definiti per completezza, accuratezza e coerenza, e al termine scrive l'esito come tag Qualità su OpenMetadata.
4. Flow 4 — Metadata: si registrano i file Bronze e le tabelle Silver in OpenMetadata, creando un'entry di lineage completa dalla fonte al layer EAV.

La quinta fase — Integrazione nel Datastore non è in capo al team di sviluppo ma al responsabile della verifica, ed è realizzata con gli strumenti descritti più avanti. Allinea il dataset trasformato al Datastore: registra le definizioni delle variabili e le etichette delle dimensioni, sincronizza i tag e le descrizioni da OpenMetadata e aggiorna le viste materializzate del catalogo, fino a rendere il dataset ricercabile e generabile come tabella Gold. La procedura concreta e la remediation dei disallineamenti sono descritte nell'Appendice A2.

Ogni pipeline risiede in una directory dedicata sotto flows/{ente}/{dataset}/ nel repository gst-maps-pipelines, contenente i quattro file flow, un manifesto di deployment prefect.yaml, un requirements.txt e un README.md che descrive la fonte, il formato e la frequenza di aggiornamento. La convenzione di naming, le utility condivise, i pattern di lavoro in team e le procedure di deployment sono descritti in dettaglio nell'Appendice A2.

Pipeline implementate

Fondazione territoriale ISTAT

La fondazione territoriale è un insieme di quattro pipeline coordinate che devono essere eseguite in sequenza prima che qualsiasi pipeline attributo possa caricare dati in Silver. Tutte le pipeline attributo risolvono gli identificatori territoriali tramite silver.territories; questa tabella — insieme alle tabelle associate di identificatori, nomi, contenimenti e relazioni — è popolata esclusivamente dalla fondazione territoriale.

flows/istat/variazioni-amministrative scarica la storia completa delle variazioni amministrative ISTAT tramite l'API REST SITUAS all'indirizzo situas-servizi.istat.it/publish. SITUAS organizza i dataset per codice pfun; la pipeline richiede dieci dataset che coprono regioni (pfun 107, 106, 108), province (pfun 113, 112, 114) e comuni (pfun 129, 98, 104, 105), ciascuno come serie storica completa dal 17 marzo 1861. Il tipo di parametro varia per dataset: alcuni accettano un intervallo di date (pdatada/pdataa), altri una singola data di riferimento (pdata). La utility condivisa situas_client.py gestisce entrambe le varianti. Le risposte grezze vengono scritte nel Bronze come CSV sotto data/bronze/istat/variazioni-amministrative/.

flows/istat/province-regioni legge i file Bronze delle variazioni e popola silver.territories, silver.territory_identifiers, silver.territory_names e silver.territory_containments per ripartizioni, regioni e province. L'ordine di elaborazione è deliberato: prima le ripartizioni, poi le regioni, poi le province, poi gli eventi di variazione (CS = Costituzione, ES = Estinzione). Gli eventi di variazione che hanno interessato il nome o i confini di una provincia vengono applicati in linea, in modo che i timestamp valid_from e valid_to su ciascun record riflettano il ciclo di vita effettivo dell'ente amministrativo.

flows/istat/comuni è la trasformazione più complessa, articolata in due fasi. La fase 1 utilizza un algoritmo union-find per identificare catene di equivalenza: quando il codice ISTAT di un comune cambia a seguito di un evento AP (Assegnazione Provinciale) o RN (Ridenominazione), i comuni collegati vengono raggruppati sotto un unico territory_id stabile. La chiave stabile è il codice catastale Belfiore (COD_CATASTO), invariante rispetto alle riorganizzazioni provinciali. La fase 2 scrive snapshot annuali di nomi, codici ISTAT e contenimenti per ciascun comune, con valid_from/valid_to derivati dall'evento di variazione corrispondente.

flows/istat/territory-corrections applica un insieme curato di correzioni che non possono essere derivate algoritmicamente dai dati sorgente SITUAS. Utilizza il codice Belfiore come chiave di ricerca ed esegue due operazioni: l'inserimento di nomi alias ADM per i comuni il cui nome ufficiale differisce dall'etichetta del registro amministrativo ADM (consentendo la risoluzione territoriale dai dataset ADM), e la correzione di anomalie di qualità dei dati note nel sorgente SITUAS (valori valid_to non aggiornati, record di contenimento mancanti).

Insieme, queste quattro pipeline stabiliscono un grafo completo e versionato temporalmente delle entità territoriali italiane dal 1861 a oggi. Ogni pipeline attributo risolve i propri identificatori sorgente in valori territory_id tramite territory_resolver prima di scrivere in silver.territory_attributes.

ADM Tabaccherie

La pipeline ADM Tabaccherie acquisisce l'elenco dei rivenditori di tabacchi autorizzati dal portale web dell'Agenzia delle Dogane e dei Monopoli (ADM). Il portale utilizza un'interfaccia JSF/PrimeFaces che richiede la gestione del token javax.faces.ViewState; la utility html_parser gestisce questa complessità in modo trasparente. Per ciascun comune, la pipeline interroga il portale, analizza la tabella HTML risultante e scrive le pagine HTML grezze nel Bronze sotto data/bronze/agenzie-dogane/tabacchi/.

Il flow di trasformazione risolve gli identificatori comunali tramite territory_resolver utilizzando la sigla provinciale e il nome del comune dalla risposta ADM, quindi scrive un record EAV per attributo (numero di punti vendita, indirizzi) in silver.territory_attributes. Questa pipeline è l'implementazione di riferimento per il pattern attributo service-count utilizzato dagli altri dataset provenienti da ADM.

Pipeline per famiglie

Le pipeline versionate si organizzano in tre gruppi.

La fondazione territoriale ISTAT, descritta sopra, è il presupposto di ogni altra pipeline e popola il grafo delle entità territoriali.

Le famiglie con orchestratore condiviso raccolgono dataset omogenei sotto un'unica logica di acquisizione e trasformazione: gli esploradati SDMX di ISTAT e ASTER ne sono gli esempi principali, dove un solo orchestratore parametrizzato copre molti dataset con poche unità di codice. È questo meccanismo a spiegare gran parte della distanza tra i 232 dataset tracciati e le circa 38 unità-pipeline versionate.

Le pipeline attributo EAV per fonte seguono il pattern descritto per ADM Tabaccherie: acquisiscono i dati di una singola fonte, risolvono gli identificatori territoriali e scrivono record EAV in silver.territory_attributes. Oltre a ISTAT (fondazione territoriale, esploradati SDMX, ASTER e altri dataset) e ad ADM, dispongono di codice pipeline le fonti AGENAS, ISPRA, ACI, AGCOM, ICCU, MUR, RFI, RGS, Ministero della Salute, Protezione Civile, Agenzie delle Dogane, Ministero del Lavoro, SISTAN e Ministero dell'Istruzione.

Delle circa 38 unità-pipeline versionate, 30 dispongono già di tutti e quattro i flow (ingestion, transform, data quality, metadata); le restanti sono in fase di completamento sui flow di qualità o di metadata.

Strumenti di verifica tecnica

La verifica tecnica delle pipeline si fonda su tre strumenti: un motore di verifica automatica, una skill operativa che lo orchestra e una dashboard di tracciamento.

Il motore di verifica (scripts/etl_status.py e il pacchetto scripts/etl_verify/) è di sola lettura e deriva i verdetti dallo stato di produzione reale: interroga il database via PostgREST, OpenMetadata, la cache dei file Bronze e il catalogo del Datastore, e non i flag di avanzamento annotati altrove. Per ogni coppia (dataset, fase) produce un verdetto su due assi — l'implementazione (yes, no, shared, n/a) e lo stato di produzione (verified o missing) — accompagnato da una stringa di evidenza. Non esiste un terzo valore «non verificabile»: se un controllo non può essere eseguito, il motore fallisce in modo esplicito anziché emettere un verdetto degradato. I cinque step verificati corrispondono alle cinque fasi: per l'ingestion la presenza dei file nel Bronze, per la transform la tabella Silver popolata, per la data quality il tag Qualità scritto su OpenMetadata, per la metadata il riferimento di lineage valorizzato, per il Datastore la configurabilità del dataset nel catalogo. La verifica copre dunque la presenza e la struttura del risultato di ciascuna fase; la correttezza semantica del dato è oggetto della validazione di contenuto descritta nel capitolo 6.

Prima di considerare attendibili i verdetti, il motore si misura contro un insieme di casi verificati a mano tramite l'opzione di self-test: se non riproduce la verità nota — per una deriva di schema o un'API irraggiungibile — il self-test fallisce e i verdetti non vanno usati.

La skill verify-etl-dataset, eseguita dal responsabile della verifica all'interno dell'ambiente Claude, codifica la procedura: esecuzione del self-test, verifica del dataset o dell'insieme di dataset, interpretazione dei verdetti, eventuale remediation e allineamento al Datastore (la quinta fase), sincronizzazione dello stato. È lo strumento con cui il responsabile coordina gli script di verifica e porta a termine l'integrazione nel Datastore.

L'avanzamento è seguito attraverso la etl-dashboard, una vista sullo stato tracciato in ClickUp, dove ogni dataset ha un sotto-task per ciascuna delle cinque fasi e un indicatore di priorità. I verdetti del motore si traducono in proposte di chiusura o riapertura dei sotto-task, applicate dopo conferma; la dashboard riflette lo stato risultante ed è la fonte autorevole dello stato per-dataset. Il quadro quantitativo di sintesi è riportato nella relazione di consegna.

Data quality framework

Il Flow 3 di ogni pipeline esegue un framework di validazione custom, articolato in due livelli che corrispondono ai layer del data lake. I controlli sul Bronze layer operano sul file grezzo tramite DuckDB e verificano formato, dimensione, struttura attesa, presenza della colonna che identifica il territorio e conteggio delle righe. I controlli sul Silver layer operano sui record EAV tramite query SQL e verificano l'esistenza della tabella di destinazione, il conteggio dei record scritti, la copertura territoriale rispetto ai comuni attivi e l'integrità referenziale verso silver.territories. Al termine della validazione il flow scrive l'esito complessivo come tag Qualità sull'entità corrispondente in OpenMetadata, rendendo lo stato di qualità di ciascun dataset consultabile direttamente dal catalogo.

La scelta di un framework custom rispetto a una libreria di validazione generica risponde a tre esigenze: leggerezza operativa, integrazione nativa con gli strumenti già adottati dalla piattaforma (DuckDB per il Bronze e OpenMetadata per la pubblicazione dell'esito) e riduzione delle dipendenze esterne. Il dettaglio dei criteri di accettazione e delle metriche di completezza, accuratezza e coerenza è trattato nel capitolo 6; questo capitolo descrive il ruolo del Flow 3 nell'architettura delle pipeline.

Accesso ai dati: il Datastore

Il layer Silver EAV è ottimizzato per la normalizzazione e la tracciabilità, non per la consultazione diretta. Per i data analyst l'accesso diretto al Silver è impraticabile: estrarre un'informazione richiede query SQL non banali sullo schema entità-attributo-valore e, per i dataset multidimensionali, la ricostruzione laboriosa delle etichette e delle dimensioni a partire dai codici. Il Datastore, raggiungibile all'indirizzo datastore.maps.gransassotech.org, risolve questo problema interponendo un'interfaccia di consultazione tra il data lake e l'utente finale.

Il Datastore offre un catalogo navigabile dei dataset disponibili, ricerca full-text, un carrello per comporre selezioni di dataset, la generazione di tabelle Gold in formato "wide" pronte all'uso, un browser del Bronze layer per ispezionare i file sorgente e il download dei dati selezionati. Le tabelle wide sono prodotte a partire dai record EAV del Silver layer: il Datastore pivota gli attributi in colonne, materializzando nel Gold layer la rappresentazione tabellare attesa dagli strumenti di analisi. In questo modo lo schema EAV resta la rappresentazione canonica e versionata dei dati, mentre il Gold layer e il Datastore ne offrono una vista denormalizzata e immediatamente utilizzabile.

Monitoring e test

Monitoring e logging

I flow run, la durata dei task e i tassi di errore sono monitorati attraverso l'interfaccia di Prefect del cluster, da cui si configurano anche le notifiche in caso di failure. I log dei worker sono disponibili nella stessa interfaccia per ogni flow run e vengono scritti nel volume mappato data/logs/workers/ sull'host.

Test

La suite di test conta 82 file e circa 920 funzioni di test. La configurazione è centralizzata in pytest.ini, che definisce tra l'altro il marker integration per distinguere i test di integrazione end-to-end dai test unitari, eseguibili in modo selettivo. La copertura riguarda sia le utility condivise sia le singole pipeline, con particolare attenzione alla logica di risoluzione territoriale e alle trasformazioni verso il layer EAV.