Cosa fa il notebook Spark in Fabric

Il notebook è lo strumento principale per la medallion architecture dentro un Lakehouse: legge dai layer inferiori, applica trasformazioni e scrive Delta table nel layer superiore. Ogni notebook è associato a un default lakehouse, che determina come si risolvono i path relativi.

Lettura: Files vs Tables

Sorgente Sintassi Note
Landing CSV/Parquet in Files spark.read.format("csv").option("header","true").load("Files/bronze/orders/") Path relativo al default lakehouse
Delta table registrata in Tables spark.read.table("silver_orders") oppure spark.sql("SELECT * FROM silver_orders") Passa dal metastore
Lakehouse diverso dal default ABFS completo: abfss://<workspace>@onelake.dfs.fabric.microsoft.com/<lakehouse>.Lakehouse/Tables/... Oppure aggiungi il lakehouse come shortcut

Sulla lettura di file, .option("inferSchema","true") costa una passata extra sui dati: in produzione definisci uno StructType esplicito.

Scrittura: saveAsTable vs save(path)

df.write.mode("overwrite").format("delta").saveAsTable("gold_sales")
df.write.mode("append").format("delta").save("Tables/gold_sales")
  • saveAsTable registra la tabella nel metastore: diventa visibile nell’explorer del Lakehouse, interrogabile dal SQL analytics endpoint e utilizzabile in Direct Lake. È la scelta di default per il layer silver/gold.
  • save(path) scrive solo i file Delta. Se il path sta sotto Tables/ Fabric la auto-registra, ma sotto Files/ resta una tabella “non gestita” invisibile all’endpoint SQL.
  • mode("overwrite") sostituisce i dati mantenendo la tabella; con .option("overwriteSchema","true") riscrive anche lo schema (serve se cambiano i tipi).
  • mode("append") aggiunge righe: se lo schema del DataFrame ha colonne nuove, aggiungi .option("mergeSchema","true") per la schema evolution, altrimenti errore.

Carichi incrementali con MERGE INTO

Per l’upsert (SCD Type 1) la strada canonica è Spark SQL:

MERGE INTO silver_customers AS t
USING staging_customers AS s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.modified_date > t.modified_date THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

Punti operativi:

  • La chiave del ON deve essere deduplicata nella sorgente: righe duplicate causano l’errore “multiple source rows matched”.
  • UPDATE SET * / INSERT * richiedono schemi compatibili; con l’autoMerge dello schema abilitato il MERGE accetta colonne nuove.
  • WHEN NOT MATCHED BY SOURCE THEN DELETE gestisce le cancellazioni (full snapshot).
  • In alternativa esiste l’API Python DeltaTable.forName(spark, "...").merge(...), equivalente funzionalmente.

Scelte di performance

Partizionamento

Usa .partitionBy("year","month") solo quando la colonna ha bassa cardinalità e filtra davvero le query. Partizionare su customer_id o su una data giornaliera con pochi record genera il classico problema dei small file: migliaia di file da pochi KB, metadata enormi, scansioni lente. Regola pratica: ogni partizione dovrebbe pesare almeno circa 1 GB. Sotto quella soglia conviene non partizionare e affidarsi a V-Order + OPTIMIZE:

OPTIMIZE gold_sales VORDER;
VACUUM gold_sales RETAIN 168 HOURS;

OPTIMIZE compatta i file (bin-compaction) e ZORDER BY (col) co-localizza i valori per il data skipping. V-Order è l’ottimizzazione di scrittura Parquet specifica di Fabric: migliora sensibilmente le letture Direct Lake.

Join e cache

  • Broadcast join: quando una dimensione è piccola (indicativamente < 100 MB), broadcast(dim_df) evita lo shuffle della fact table. Adaptive Query Execution lo applica spesso da solo, ma forzarlo su join ricorrenti è sicuro.
  • Cache: df.cache() conviene solo se il DataFrame viene riusato più volte nella stessa sessione. Su una pipeline lineare (leggi → trasforma → scrivi) la cache aggiunge solo pressione di memoria.

Impostazioni Spark del workspace

Sotto Workspace settings → Data Engineering/Science → Spark settings:

Impostazione Effetto
Starter pool Nodi pre-riscaldati: avvio sessione in pochi secondi. Default, ottimo per sviluppo.
Custom pool Node size e autoscale personalizzati, ma avvio più lento (provisioning). Serve per librerie custom o memoria elevata.
High concurrency Più notebook condividono una sessione Spark: meno CU consumate e nessuna attesa di avvio ripetuta.
Native execution engine Runtime vettorizzato: query più veloci a parità di nodi, quindi meno CU-secondi.
Environment Artefatto che porta librerie e configurazioni Spark; se allegato può impedire l’uso dello starter pool.

Il consumo si misura in CU proporzionali a core × durata: sessioni che restano appese sprecano capacity, quindi tieni basso il timeout di inattività.

Trappole tipiche d’esame

  • Scenario: la tabella scritta dal notebook non compare nel SQL analytics endpoint né in Direct Lake → Risposta: è stata scritta con save() sotto Files/; riscrivila con saveAsTable() (o sotto Tables/) perché venga registrata nel metastore.
  • Scenario: un append fallisce perché la sorgente ha una colonna nuova, e non si vuole ricaricare tutto → Risposta: .option("mergeSchema","true"), non overwriteSchema (che riscriverebbe i dati esistenti).
  • Scenario: dopo il partizionamento per order_date le query sono più lente e i file sono migliaia → Risposta: rimuovi il partizionamento a granularità troppo fine e usa OPTIMIZE ... VORDER con eventuale ZORDER.
  • Scenario: il MERGE fallisce con “multiple source rows matched a target row” → Risposta: deduplica la staging sulla chiave di join (es. row_number() sull’ultima modified_date) prima del MERGE.
  • Scenario: molti notebook brevi in sequenza e il tempo va tutto in avvio sessione → Risposta: mantieni lo starter pool e abilita high concurrency, non un custom pool.