{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 17 · Modificare tabelle\n\nApplica modifiche a tabelle Delta temporanee e verifica che cosa succede quando ripeti un lotto. Usa una nuova sessione CPU su Kaggle con Java 17 e Internet abilitato. Esegui `%pip install \"pyspark==4.0.0\" \"delta-spark==4.0.0\"` prima di importare Spark e riavvia il kernel se serve. Spark scarica le dipendenze JVM di Delta all’avvio. Tutte le destinazioni sono in una directory temporanea.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\nfrom uuid import uuid4\n\nfrom delta import configure_spark_with_delta_pip\nfrom delta.tables import DeltaTable\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nscratch = TemporaryDirectory(prefix=\"spark-nb17-\")\nroot = Path(scratch.name)\nbuilder = (\n    SparkSession.builder.master(\"local[2]\").appName(\"17-modify-tables\")\n    .config(\"spark.sql.extensions\", \"io.delta.sql.DeltaSparkSessionExtension\")\n    .config(\"spark.sql.catalog.spark_catalog\", \"org.apache.spark.sql.delta.catalog.DeltaCatalog\")\n    .config(\"spark.sql.shuffle.partitions\", \"2\")\n)\nspark = configure_spark_with_delta_pip(builder).getOrCreate()\n\ndef delta_path(name):\n    return str(root / name)\n\nschema = \"id INT, status STRING\"\npath = delta_path(\"orders\")\ninitial = spark.createDataFrame([(1, \"new\"), (2, \"new\")], schema)\ninitial.write.format(\"delta\").save(path)\ntarget = DeltaTable.forPath(spark, path)\nchanges = spark.createDataFrame([(1, \"paid\"), (3, \"new\")], schema)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Formato di tabella\n\nDelta definisce un protocollo su file di dati e un registro delle transazioni. Il lettore usa il registro per individuare i file che appartengono a uno snapshot.\n\n## 2. Registro delle transazioni\n\nOgni modifica confermata incrementa la versione della tabella. Esamina la cronologia e la directory `_delta_log` della tabella di prova.\n\n## 3. Scrittura idempotente\n\nRipetere un lotto è idempotente quando lascia lo stesso stato logico. Una scrittura ripetuta può comunque creare un altro commit.\n\n## 4. Upsert\n\nUn upsert aggiorna le chiavi che trovano corrispondenza e inserisce quelle nuove. Qui la sorgente ha un record per ID.\n\n## 5. Append alla cieca\n\nAccodare tutte le righe senza controllare le chiavi può duplicare i record di business quando ripeti il lotto.\n\n## 6. Upsert su tabelle grandi\n\nUn merge può riscrivere i file interessati. Limita l’input al lotto pertinente, deduplica le chiavi e usa predicati di partizione solo se preservano tutte le corrispondenze valide. Esamina le metriche dell’operazione.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert (Path(path) / \"_delta_log\").is_dir()\ntarget.history(1).select(\"version\", \"operation\").show()\nappend_path = delta_path(\"append\")\nfor _ in range(2):\n    changes.write.format(\"delta\").mode(\"append\").save(append_path)\nassert spark.read.format(\"delta\").load(append_path).count() == 4",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. MERGE INTO\n\nMERGE in SQL può aggiornare le righe corrispondenti e inserire quelle senza corrispondenza. Ripeti lo stesso input e verifica il risultato.\n\n## 8. Cattura delle modifiche\n\nIl CDC registra operazioni come inserimenti, aggiornamenti ed eliminazioni. Chi le applica ha bisogno di una chiave e di una regola di ordinamento per risolvere più eventi della stessa chiave. Qui il lotto ha già chiavi univoche e una colonna per l’operazione.\n\n## 9. MERGE sulle righe assenti dalla sorgente\n\n`WHEN NOT MATCHED BY SOURCE` può aggiornare o eliminare righe della destinazione assenti da uno snapshot completo. Un lotto CDC parziale non basta per interpretare le assenze come eliminazioni.\n\n## 10. Chiavi duplicate nella sorgente\n\nPiù righe sorgente che corrispondono alla stessa destinazione possono rendere ambiguo l’aggiornamento. Più righe senza corrispondenza possono anche inserire chiavi duplicate. Verifica l’unicità prima del merge.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "suffix = uuid4().hex[:8]\nchanges_view = \"nb17_changes_\" + suffix\nchanges.createOrReplaceTempView(changes_view)\n\ndef require_unique(df):\n    assert df.filter(\"id IS NULL\").count() == 0\n    assert df.groupBy(\"id\").count().filter(\"count > 1\").count() == 0\n\nrequire_unique(changes)\nfor _ in range(2):\n    spark.sql(f'''\n        MERGE INTO delta.`{path}` AS t USING {changes_view} AS s ON t.id = s.id\n        WHEN MATCHED THEN UPDATE SET t.status = s.status\n        WHEN NOT MATCHED THEN INSERT (id, status) VALUES (s.id, s.status)\n    ''')\nassert {r.id: r.status for r in target.toDF().collect()} == {1: \"paid\", 2: \"new\", 3: \"new\"}\n\nsnapshot_path = delta_path(\"snapshot\")\ninitial.write.format(\"delta\").save(snapshot_path)\nspark.sql(f'''\n    MERGE INTO delta.`{snapshot_path}` t USING {changes_view} s ON t.id = s.id\n    WHEN MATCHED THEN UPDATE SET t.status = s.status\n    WHEN NOT MATCHED THEN INSERT (id, status) VALUES (s.id, s.status)\n    WHEN NOT MATCHED BY SOURCE THEN DELETE\n''')\nassert {r.id for r in spark.read.format(\"delta\").load(snapshot_path).collect()} == {1, 3}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. MERGE con l’API Delta Lake\n\n`DeltaTable.merge()` costruisce la stessa operazione in Python. Qui un’eliminazione CDC esplicita rimuove l’ID 2 e un aggiornamento modifica l’ID 3.\n\n## 12. MERGE con l’API DataFrame\n\nSpark 4 aggiunge `DataFrame.mergeInto()`. Opera su una tabella con nome e richiede un provider compatibile. Qui la funzione è definita senza invocarla. Chiamala solo su una destinazione di prova compatibile. Per i dati di prova della lezione si usa l’API Delta Lake appena vista.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "cdc = spark.createDataFrame([(2, None, \"delete\"), (3, \"paid\", \"upsert\")], schema + \", op STRING\")\nrequire_unique(cdc)\n(\n    target.alias(\"t\").merge(cdc.alias(\"s\"), \"t.id = s.id\")\n    .whenMatchedDelete(\"s.op = 'delete'\")\n    .whenMatchedUpdate(\"s.op = 'upsert'\", set={\"status\": \"s.status\"})\n    .whenNotMatchedInsert(\"s.op = 'upsert'\", values={\"id\": \"s.id\", \"status\": \"s.status\"})\n    .execute()\n)\nassert {r.id: r.status for r in target.toDF().collect()} == {1: \"paid\", 3: \"paid\"}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "Il frammento seguente presume una tabella di prova compatibile e con nome, con colonne id e status.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def merge_with_dataframe_api(source, table_name):\n    target_name = table_name.rsplit(\".\", 1)[-1]\n    (\n        source.alias(\"s\").mergeInto(table_name, F.col(\"s.id\") == F.col(f\"{target_name}.id\"))\n        .whenMatched().update({\"status\": F.col(\"s.status\")})\n        .whenNotMatched().insert({\"id\": F.col(\"s.id\"), \"status\": F.col(\"s.status\")})\n        .merge()\n    )",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Sovrascrittura dinamica delle partizioni\n\nLa sovrascrittura dinamica sostituisce solo le partizioni di destinazione rappresentate nei dati in ingresso. Una partizione inattesa può causare sostituzioni indesiderate.\n\n## 14. Sovrascrittura statica delle altre partizioni\n\nIn questa sovrascrittura statica senza restrizioni si sostituisce l’intera tabella. Prova entrambe le modalità su copie temporanee separate e confronta le partizioni conservate.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "partitioned = spark.createDataFrame([(1, \"A\"), (2, \"B\")], \"id INT, zone STRING\")\nreplacement = spark.createDataFrame([(3, \"A\")], partitioned.schema)\nfor mode, expected in [(\"dynamic\", {2, 3}), (\"static\", {3})]:\n    destination = delta_path(mode)\n    partitioned.write.format(\"delta\").partitionBy(\"zone\").save(destination)\n    replacement.write.format(\"delta\").mode(\"overwrite\").option(\"partitionOverwriteMode\", mode).save(destination)\n    assert {r.id for r in spark.read.format(\"delta\").load(destination).collect()} == expected",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Dimensione a variazione lenta di tipo 2\n\nIl tipo 2 mantiene una riga storica per intervallo di validità. Usa intervalli semiaperti affinché un fatto al confine corrisponda alla nuova versione. Qui si costruisce una piccola cronologia da modifiche ordinate.\n\n## 16. Dimensione a variazione lenta di tipo 1\n\nIl tipo 1 conserva solo il valore più recente. Unendo per la sola chiave della dimensione, anche i fatti precedenti vedranno l’etichetta attuale.\n\n## 17. Join con una dimensione di tipo 2\n\nAbbina sia la chiave dell’entità sia il tempo del fatto. Verifica che gli intervalli non si sovrappongano e che ogni fatto atteso abbia una sola corrispondenza.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import Window\n\nlabels = spark.createDataFrame([(1, 1, \"basic\"), (1, 5, \"plus\")], \"id INT, start_day INT, tier STRING\")\nby_time = Window.partitionBy(\"id\").orderBy(\"start_day\")\nhistory = labels.withColumn(\"end_day\", F.lead(\"start_day\").over(by_time))\ncurrent = history.filter(\"end_day IS NULL\")\nassert current.first().tier == \"plus\"\nfacts = spark.createDataFrame([(1, 4), (1, 5)], \"id INT, day INT\")\nf, h = facts.alias(\"f\"), history.alias(\"h\")\nas_of = f.join(h, (f.id == h.id) & (f.day >= h.start_day) & (h.end_day.isNull() | (f.day < h.end_day)))\nassert [(r.day, r.tier) for r in as_of.select(\"f.day\", \"h.tier\").orderBy(\"day\").collect()] == [(4, \"basic\"), (5, \"plus\")]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 18. Caricamento incrementale\n\nLeggi i record dopo l’ultimo limite salvato e fino a un estremo di lotto fisso. Qui gli ID di sequenza danno un ordine totale. Con timestamp uguali, usa uno spareggio o una sovrapposizione sicura in caso di ripetizione.\n\n## 19. Salvare il massimo dopo la scrittura\n\nAvanza il limite solo dopo che la scrittura a destinazione è riuscita. Un arresto tra i due passaggi può causare la ripetizione del lotto, quindi rendi sicuro il nuovo tentativo.\n\n## 20. Rendere idempotente un job di scrittura\n\nDelta può riconoscere append duplicati tramite ID applicazione e versione della transazione. Riusa la coppia per ritentare lo stesso lotto logico. In un job reale usa un ID applicazione stabile e versioni crescenti dei lotti.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "feed = spark.createDataFrame([(1, \"a\"), (2, \"b\"), (3, \"c\")], \"sequence_id INT, value STRING\")\nlast_committed = 1\nbatch_end = 3\nbatch = feed.filter((F.col(\"sequence_id\") > last_committed) & (F.col(\"sequence_id\") <= batch_end))\nincremental_path = delta_path(\"incremental\")\nfor _ in range(2):\n    (\n        batch.write.format(\"delta\").mode(\"append\")\n        .option(\"txnAppId\", \"nb17-incremental-\" + suffix)\n        .option(\"txnVersion\", batch_end).save(incremental_path)\n    )\nassert spark.read.format(\"delta\").load(incremental_path).count() == 2\nlast_committed = batch.agg(F.max(\"sequence_id\")).first()[0]\nassert last_committed == 3",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 21. Viaggio nel tempo\n\nUsa `versionAsOf` per leggere uno snapshot precedente finché il registro e i file necessari sono disponibili. La versione zero contiene ancora i due ordini originali.\n\n## 22. VACUUM\n\nVACUUM elimina i file non più referenziati quando soddisfano i criteri di conservazione. Qui si usa una simulazione con la conservazione predefinita. Eliminare file vecchi può rendere illeggibili snapshot precedenti.\n\n## 23. Scrivere e leggere in formato Delta\n\nUsa `format(\"delta\")` per creare e leggere un dataset Delta per percorso. Il lettore seleziona uno snapshot coerente.\n\n## 24. Compattare i file\n\n`OPTIMIZE` può combinare file piccoli in file più grandi. Verifica che le righe logiche restino uguali. Conteggi dei file e metriche dipendono dalla disposizione iniziale.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "original = spark.read.format(\"delta\").option(\"versionAsOf\", 0).load(path)\nassert {r.id: r.status for r in original.collect()} == {1: \"new\", 2: \"new\"}\nspark.sql(f\"VACUUM delta.`{path}` DRY RUN\").show(truncate=False)\n\ncompact_path = delta_path(\"compact\")\nfor start in (0, 2, 4):\n    spark.range(start, start + 2).coalesce(1).write.format(\"delta\").mode(\"append\").save(compact_path)\ncompaction = DeltaTable.forPath(spark, compact_path)\ncompaction.optimize().executeCompaction().show(truncate=False)\nassert {r.id for r in spark.read.format(\"delta\").load(compact_path).collect()} == set(range(6))\n\nspark.catalog.dropTempView(changes_view)\nscratch.cleanup()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché è rischioso eliminare per assenza usando un lotto CDC parziale?\n2. Quando deve avanzare il limite?\n3. Quali colonne determinano se un fatto corrisponde a una versione di tipo 2?\n\n### Risposte\n\n1. Una chiave assente può indicare che nel lotto non c’è stata alcuna modifica.\n2. Dopo un commit riuscito a destinazione. La scrittura deve poter essere ripetuta se l’aggiornamento del limite fallisce.\n3. La chiave dell’entità e la data del fatto rispetto all’intervallo di validità.",
      "id": "cell-19"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}