Il modello micro-batch

Spark Structured Streaming tratta uno stream come una tabella append-only che cresce nel temp: ogni nuovo dato è una riga aggiunta a una unbounded table. Il motore non elabora record uno a uno (tranne il caso opzionale continuous processing, raro all’esame), ma li raggruppa in micro-batch. A ogni trigger Spark legge i nuovi dati dalla sorgente, esegue lo stesso piano logico di una query batch e scrive il risultato nel sink. Il grande vantaggio: le stesse API DataFrame valgono per batch e streaming, quindi puoi riusare la logica di trasformazione senza riscriverla.

readStream, writeStream e output mode

Si legge con spark.readStream (formati tipici: Kafka, Event Hubs via connettore, Auto Loader su ADLS) e si scrive con writeStream. La scelta cruciale è l’output mode, che definisce cosa viene emesso al sink a ogni batch:

  • append: emette solo le righe nuove che non cambieranno più. È il default e l’unico valido per query senza aggregazione. Con aggregazioni richiede un watermark, perché solo così Spark sa quando una finestra è “chiusa” e la riga è definitiva.
  • update: emette solo le righe la cui aggregazione è cambiata nel batch corrente. Ideale per aggregazioni continue verso sink che supportano upsert.
  • complete: riscrive l’intera tabella dei risultati ogni volta. Utile per aggregazioni globali piccole (es. conteggi totali), ma costoso e non scalabile: lo stato cresce senza limiti.
(df.writeStream
   .format("delta")
   .outputMode("append")
   .option("checkpointLocation", "/mnt/chk/bronze_orders")
   .trigger(availableNow=True)
   .toTable("bronze.orders"))

Checkpointing e semantica exactly-once

Il checkpointLocation è obbligatorio in produzione: contiene gli offset letti dalla sorgente e lo stato delle aggregazioni. Al riavvio Spark riparte esattamente dall’ultimo offset committato. Combinando checkpoint (offset tracking + write-ahead log) con un sink idempotente/transazionale come Delta Lake, si ottiene la semantica exactly-once end-to-end. Punto chiave d’esame: ogni query deve avere il proprio checkpoint dedicato; condividerlo tra query diverse corrompe lo stato.

Trigger

Il trigger decide la cadenza dei micro-batch:

  • default (nessun trigger): appena finito un batch ne parte un altro, minimizzando la latenza; il cluster resta acceso.
  • fixed interval (processingTime="1 minute"): batch a intervalli regolari, per bilanciare latenza e costo.
  • availableNow: processa tutti i dati disponibili in uno o più batch e poi si ferma. È il pattern moderno (sostituisce once) per pipeline incrementali schedulate: paghi il cluster solo durante l’esecuzione, mantenendo comunque lo stato via checkpoint tra un run e l’altro.

Stateful aggregation: windowing e watermark

Per aggregare nel tempo si usa window() su una colonna di event-time (non processing-time): tumbling o sliding window. Il problema sono i dati in ritardo: senza limiti lo stato crescerebbe all’infinito. Il watermark (withWatermark("eventTime", "10 minutes")) dichiara quanto tardi può arrivare un evento: oltre quella soglia Spark scarta il record e può chiudere ed emettere le finestre più vecchie, liberando stato. È il watermark che rende praticabile l’append mode sulle aggregazioni.

(events.withWatermark("eventTime", "10 minutes")
       .groupBy(window("eventTime", "5 minutes"), "deviceId")
       .count())

Delta e architettura medallion

Structured Streaming è la spina dorsale della medallion architecture su Databricks: streaming da sorgente a Bronze (raw, spesso via Auto Loader), poi trasformazioni streaming Bronze→Silver (pulizia, dedup) e Silver→Gold (aggregati business). Delta abilita MERGE, time travel e letture incrementali (Change Data Feed), rendendo ogni layer sia streaming che batch.

Structured Streaming vs Stream Analytics

  • Azure Stream Analytics: linguaggio SQL dichiarativo, serverless, setup rapidissimo, ottimo per integrazione nativa Event Hubs/IoT Hub e Power BI. Meno flessibile su logica custom.
  • Structured Streaming: pieno controllo via codice (Python/Scala), librerie ML, join complessi, integrazione Delta/medallion nativa. Richiede un cluster Spark e più competenze.

Regola pratica: SQL semplice e time-to-market → Stream Analytics; trasformazioni complesse, lakehouse Delta e riuso della logica batch → Structured Streaming.

Trappole tipiche d’esame

  • Aggregazione + append mode senza watermark → la query non emette mai righe (o va in errore): serve withWatermark per chiudere le finestre.
  • Serve processare l’arretrato una volta e spegnere il cluster → trigger availableNow, non default né fixed interval.
  • Due query condividono lo stesso checkpointLocation → stato corrotto/exactly-once perso: ogni sink vuole un checkpoint dedicato.
  • Vuoi solo i valori aggiornati verso un sink con upsert → update mode; complete solo per aggregati globali piccoli.
  • Latenza tollerabile ma serve minimo costo compute → fixed interval o availableNow; il trigger default tiene il cluster sempre occupato.
  • Watermark impostato su processing-time invece che event-time → conteggi finestrati errati con dati out-of-order: usa sempre la colonna di event-time.