17 · Modificare tabelle

Applica 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.

from pathlib import Path
from tempfile import TemporaryDirectory
from uuid import uuid4

from delta import configure_spark_with_delta_pip
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

scratch = TemporaryDirectory(prefix="spark-nb17-")
root = Path(scratch.name)
builder = (
    SparkSession.builder.master("local[2]").appName("17-modify-tables")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .config("spark.sql.shuffle.partitions", "2")
)
spark = configure_spark_with_delta_pip(builder).getOrCreate()

def delta_path(name):
    return str(root / name)

schema = "id INT, status STRING"
path = delta_path("orders")
initial = spark.createDataFrame([(1, "new"), (2, "new")], schema)
initial.write.format("delta").save(path)
target = DeltaTable.forPath(spark, path)
changes = spark.createDataFrame([(1, "paid"), (3, "new")], schema)

1. Formato di tabella

Delta 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.

2. Registro delle transazioni

Ogni modifica confermata incrementa la versione della tabella. Esamina la cronologia e la directory _delta_log della tabella di prova.

3. Scrittura idempotente

Ripetere un lotto è idempotente quando lascia lo stesso stato logico. Una scrittura ripetuta può comunque creare un altro commit.

4. Upsert

Un upsert aggiorna le chiavi che trovano corrispondenza e inserisce quelle nuove. Qui la sorgente ha un record per ID.

5. Append alla cieca

Accodare tutte le righe senza controllare le chiavi può duplicare i record di business quando ripeti il lotto.

6. Upsert su tabelle grandi

Un 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.

assert (Path(path) / "_delta_log").is_dir()
target.history(1).select("version", "operation").show()
append_path = delta_path("append")
for _ in range(2):
    changes.write.format("delta").mode("append").save(append_path)
assert spark.read.format("delta").load(append_path).count() == 4

7. MERGE INTO

MERGE in SQL può aggiornare le righe corrispondenti e inserire quelle senza corrispondenza. Ripeti lo stesso input e verifica il risultato.

8. Cattura delle modifiche

Il 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.

9. MERGE sulle righe assenti dalla sorgente

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.

10. Chiavi duplicate nella sorgente

Più 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.

suffix = uuid4().hex[:8]
changes_view = "nb17_changes_" + suffix
changes.createOrReplaceTempView(changes_view)

def require_unique(df):
    assert df.filter("id IS NULL").count() == 0
    assert df.groupBy("id").count().filter("count > 1").count() == 0

require_unique(changes)
for _ in range(2):
    spark.sql(f'''
        MERGE INTO delta.`{path}` AS t USING {changes_view} AS s ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET t.status = s.status
        WHEN NOT MATCHED THEN INSERT (id, status) VALUES (s.id, s.status)
    ''')
assert {r.id: r.status for r in target.toDF().collect()} == {1: "paid", 2: "new", 3: "new"}

snapshot_path = delta_path("snapshot")
initial.write.format("delta").save(snapshot_path)
spark.sql(f'''
    MERGE INTO delta.`{snapshot_path}` t USING {changes_view} s ON t.id = s.id
    WHEN MATCHED THEN UPDATE SET t.status = s.status
    WHEN NOT MATCHED THEN INSERT (id, status) VALUES (s.id, s.status)
    WHEN NOT MATCHED BY SOURCE THEN DELETE
''')
assert {r.id for r in spark.read.format("delta").load(snapshot_path).collect()} == {1, 3}

11. MERGE con l’API Delta Lake

DeltaTable.merge() costruisce la stessa operazione in Python. Qui un’eliminazione CDC esplicita rimuove l’ID 2 e un aggiornamento modifica l’ID 3.

12. MERGE con l’API DataFrame

Spark 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.

cdc = spark.createDataFrame([(2, None, "delete"), (3, "paid", "upsert")], schema + ", op STRING")
require_unique(cdc)
(
    target.alias("t").merge(cdc.alias("s"), "t.id = s.id")
    .whenMatchedDelete("s.op = 'delete'")
    .whenMatchedUpdate("s.op = 'upsert'", set={"status": "s.status"})
    .whenNotMatchedInsert("s.op = 'upsert'", values={"id": "s.id", "status": "s.status"})
    .execute()
)
assert {r.id: r.status for r in target.toDF().collect()} == {1: "paid", 3: "paid"}

Il frammento seguente presume una tabella di prova compatibile e con nome, con colonne id e status.

def merge_with_dataframe_api(source, table_name):
    target_name = table_name.rsplit(".", 1)[-1]
    (
        source.alias("s").mergeInto(table_name, F.col("s.id") == F.col(f"{target_name}.id"))
        .whenMatched().update({"status": F.col("s.status")})
        .whenNotMatched().insert({"id": F.col("s.id"), "status": F.col("s.status")})
        .merge()
    )

13. Sovrascrittura dinamica delle partizioni

La sovrascrittura dinamica sostituisce solo le partizioni di destinazione rappresentate nei dati in ingresso. Una partizione inattesa può causare sostituzioni indesiderate.

14. Sovrascrittura statica delle altre partizioni

In questa sovrascrittura statica senza restrizioni si sostituisce l’intera tabella. Prova entrambe le modalità su copie temporanee separate e confronta le partizioni conservate.

partitioned = spark.createDataFrame([(1, "A"), (2, "B")], "id INT, zone STRING")
replacement = spark.createDataFrame([(3, "A")], partitioned.schema)
for mode, expected in [("dynamic", {2, 3}), ("static", {3})]:
    destination = delta_path(mode)
    partitioned.write.format("delta").partitionBy("zone").save(destination)
    replacement.write.format("delta").mode("overwrite").option("partitionOverwriteMode", mode).save(destination)
    assert {r.id for r in spark.read.format("delta").load(destination).collect()} == expected

15. Dimensione a variazione lenta di tipo 2

Il 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.

16. Dimensione a variazione lenta di tipo 1

Il tipo 1 conserva solo il valore più recente. Unendo per la sola chiave della dimensione, anche i fatti precedenti vedranno l’etichetta attuale.

17. Join con una dimensione di tipo 2

Abbina 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.

from pyspark.sql import Window

labels = spark.createDataFrame([(1, 1, "basic"), (1, 5, "plus")], "id INT, start_day INT, tier STRING")
by_time = Window.partitionBy("id").orderBy("start_day")
history = labels.withColumn("end_day", F.lead("start_day").over(by_time))
current = history.filter("end_day IS NULL")
assert current.first().tier == "plus"
facts = spark.createDataFrame([(1, 4), (1, 5)], "id INT, day INT")
f, h = facts.alias("f"), history.alias("h")
as_of = f.join(h, (f.id == h.id) & (f.day >= h.start_day) & (h.end_day.isNull() | (f.day < h.end_day)))
assert [(r.day, r.tier) for r in as_of.select("f.day", "h.tier").orderBy("day").collect()] == [(4, "basic"), (5, "plus")]

18. Caricamento incrementale

Leggi 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.

19. Salvare il massimo dopo la scrittura

Avanza 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.

20. Rendere idempotente un job di scrittura

Delta 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.

feed = spark.createDataFrame([(1, "a"), (2, "b"), (3, "c")], "sequence_id INT, value STRING")
last_committed = 1
batch_end = 3
batch = feed.filter((F.col("sequence_id") > last_committed) & (F.col("sequence_id") <= batch_end))
incremental_path = delta_path("incremental")
for _ in range(2):
    (
        batch.write.format("delta").mode("append")
        .option("txnAppId", "nb17-incremental-" + suffix)
        .option("txnVersion", batch_end).save(incremental_path)
    )
assert spark.read.format("delta").load(incremental_path).count() == 2
last_committed = batch.agg(F.max("sequence_id")).first()[0]
assert last_committed == 3

21. Viaggio nel tempo

Usa versionAsOf per leggere uno snapshot precedente finché il registro e i file necessari sono disponibili. La versione zero contiene ancora i due ordini originali.

22. VACUUM

VACUUM 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.

23. Scrivere e leggere in formato Delta

Usa format("delta") per creare e leggere un dataset Delta per percorso. Il lettore seleziona uno snapshot coerente.

24. Compattare i file

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.

original = spark.read.format("delta").option("versionAsOf", 0).load(path)
assert {r.id: r.status for r in original.collect()} == {1: "new", 2: "new"}
spark.sql(f"VACUUM delta.`{path}` DRY RUN").show(truncate=False)

compact_path = delta_path("compact")
for start in (0, 2, 4):
    spark.range(start, start + 2).coalesce(1).write.format("delta").mode("append").save(compact_path)
compaction = DeltaTable.forPath(spark, compact_path)
compaction.optimize().executeCompaction().show(truncate=False)
assert {r.id for r in spark.read.format("delta").load(compact_path).collect()} == set(range(6))

spark.catalog.dropTempView(changes_view)
scratch.cleanup()

Verifica la comprensione

1. Perché è rischioso eliminare per assenza usando un lotto CDC parziale?
2. Quando deve avanzare il limite?
3. Quali colonne determinano se un fatto corrisponde a una versione di tipo 2?

Risposte

1. Una chiave assente può indicare che nel lotto non c’è stata alcuna modifica.
2. Dopo un commit riuscito a destinazione. La scrittura deve poter essere ripetuta se l’aggiornamento del limite fallisce.
3. La chiave dell’entità e la data del fatto rispetto all’intervallo di validità.