Auto Loader: ingestion incrementale su cloud storage

Auto Loader è la sorgente cloudFiles di Structured Streaming su Databricks che ingerisce in modo incrementale i file man mano che arrivano su un cloud storage (ADLS Gen2, Blob, S3). A differenza di una lettura batch che rilegge l’intera directory, Auto Loader tiene traccia dei file già processati tramite il checkpoint e ingerisce solo i nuovi, garantendo semantica exactly-once.

Discovery dei file: directory listing vs file notification

La modalità di scoperta dei nuovi file è una scelta architetturale ricorrente:

  • Directory listing (default): elenca periodicamente i file. Semplice, nessuna risorsa extra, ma diventa lento e costoso su directory con milioni di file.
  • File notification: usa i servizi di eventi del cloud (in Azure Event Grid + Queue Storage) per essere notificato dei nuovi blob. Scala molto meglio ad alto throughput e su directory profonde, al costo di permessi per creare le risorse di notifica. Si attiva con cloudFiles.useNotifications = true.

Regola pratica: pochi file / bassa frequenza → directory listing; milioni di file / alto volume → file notification.

Schema inference ed evolution

Auto Loader inferisce lo schema al primo run e lo persiste nella cloudFiles.schemaLocation. All’arrivo di nuove colonne applica la schema evolution secondo cloudFiles.schemaEvolutionMode:

  • addNewColumns (default): aggiunge le colonne e fa fallire lo stream, che riparte con lo schema aggiornato (il restart è previsto, non un errore).
  • rescue: non fa fallire; incanala i dati non conformi.
  • none / failOnNewColumns: comportamenti più rigidi.

La colonna _rescued_data cattura campi che non combaciano con lo schema (tipi incompatibili, colonne extra): è il paracadute per non perdere dati durante i cambi di sorgente.

df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint_path)
    .load(source_path))

Delta Live Tables: pipeline dichiarative

Delta Live Tables (DLT) — oggi parte di Lakeflow Declarative Pipelines — permette di definire in modo dichiarativo le trasformazioni: dichiari le tabelle/viste risultanti e il framework gestisce orchestrazione, dipendenze, retry, checkpoint e cluster. Il DAG delle dipendenze è dedotto automaticamente dai riferimenti tra dataset.

Streaming tables vs materialized views

  • Streaming table: definita su una sorgente streaming (tipicamente Auto Loader), elabora ogni record una sola volta. Ideale per ingestion append-only (layer bronze). Mantiene stato e checkpoint.
  • Materialized view (già “live table”): ricalcolata per riflettere lo stato corrente dell’intera sorgente. Adatta a trasformazioni e aggregazioni (silver/gold) dove serve coerenza sul dato completo.

Scelta: dati continui da appendere → streaming table; aggregazioni che devono restare consistenti con tutta la sorgente → materialized view.

Expectations per la data quality

Le expectations sono vincoli di qualità dichiarativi che decidono cosa fare dei record non validi:

  • expect (warn): registra la violazione nelle metriche, mantiene il record.
  • expect_or_drop: scarta i record che violano.
  • expect_or_fail: fa fallire la pipeline.

Le metriche confluiscono nell’event log della pipeline, che fornisce anche lineage e monitoraggio nativi.

DLT vs job Spark scritti a mano

  • Con job Spark manuali gestisci tu checkpoint, ordine delle dipendenze, retry, evoluzione schema e cluster.
  • Con DLT dichiari le tabelle; qualità dati, lineage e orchestrazione sono integrati.
  • Scegli DLT per ridurre il codice operativo e avere data quality/lineage out-of-the-box; scegli job manuali quando serve controllo estremo o logica non esprimibile in modo dichiarativo.

Trappole tipiche d’esame

  • Directory con milioni di file e listing lento → non aumentare i cluster: passa a file notification mode (Event Grid + Queue), non insistere sul directory listing.
  • Serve tracciare i file già ingeriti senza rielaborarli → usa Auto Loader con checkpoint/schemaLocation dedicati; puntare due stream diversi allo stesso checkpoint, o cancellarlo, causa riprocessamento o perdita di stato.
  • Arrivano nuove colonne e lo stream si ferma → con addNewColumns il restart è atteso; se il requisito è “non fermarsi mai e non perdere dati”, scegli rescue mode con _rescued_data.
  • Ingestion append-only continua in DLT → streaming table (una sola elaborazione per record), non materialized view (che ricalcola tutto).
  • “Scartare le righe non valide ma continuare la pipeline” → expect_or_drop; expect conserva i record, expect_or_fail blocca l’esecuzione.
  • Voler eliminare orchestrazione, retry e lineage manuali → Delta Live Tables, non un job Spark custom schedulato.