13 · Controllare chiavi e duplicati

Verifica l’identità delle righe, scegli in modo deterministico quale conservare e prevedi i conteggi del join. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa con %pip install "pyspark>=4.0,<4.1" in una cella di preparazione. I dati sono nel notebook.

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[2]").appName("13-check-keys").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
updates = spark.createDataFrame(
    [(101, 1, 1, 10, 2), (101, 1, 1, 10, 2), (101, 1, 2, 11, 3),
     (101, 1, 2, 12, 4), (101, 2, 1, 13, 1), (102, 1, 1, 14, 2),
     (999, 1, 1, 15, 5), (None, 1, 1, 16, 1)],
    "order_id INT, line_id INT, revision INT, event_id INT, units INT",
)
orders = spark.createDataFrame([(101,), (102,), (103,)], "order_id INT")
line_key = ["order_id", "line_id"]

1. Duplicato esatto

Due righe sono duplicati esatti quando ogni colonna ha lo stesso valore. Le due copie dell’evento 10 coincidono in tutti i campi.

2. Chiave primaria

Una chiave primaria identifica una riga e deve essere univoca e presente. La tabella delle intestazioni usa order_id. Verifica entrambe le proprietà nel DataFrame.

3. Chiave composta

Una riga d’ordine è identificata da order_id e line_id insieme. Un ID di riga può ripetersi tra ordini diversi. La tabella degli aggiornamenti ha una grana più fine perché una riga può avere più revisioni.

4. Chiave esterna

L’order_id della riga fa riferimento alla tabella delle intestazioni. I valori possono ripetersi perché un ordine può avere più righe.

5. Orfano

L’ordine 999 ha una chiave esterna presente ma nessuna intestazione. Tieni traccia a parte delle chiavi mancanti. L’uguaglianza ordinaria del join non abbina chiavi null.

assert orders.filter("order_id IS NULL").count() == 0
assert orders.select("order_id").distinct().count() == orders.count()
orphans = updates.filter("order_id IS NOT NULL").join(orders, "order_id", "left_anti")
assert {r.order_id for r in orphans.collect()} == {999}
assert updates.filter("order_id IS NULL").count() == 1

6. Eliminare righe duplicate

Decidi quali righe considerare duplicate prima di eliminarle. Acquisizioni ripetute di uno stesso evento e revisioni diverse di una riga d’ordine richiedono regole diverse.

7. Eliminare i duplicati esatti

distinct() e dropDuplicates() senza un sottoinsieme mantengono una copia di ogni riga completa.

8. distinct come trasformazione larga

Le copie possono trovarsi in partizioni diverse. Spark in genere usa uno shuffle per riunire le righe uguali. Cerca uno scambio nel piano fisico.

9. Trovare i duplicati

Raggruppa per le colonne che definiscono l’identità, conta le righe e conserva i gruppi con più di una riga.

exact_groups = updates.groupBy(*updates.columns).count().filter("count > 1")
assert [(r.event_id, r["count"]) for r in exact_groups.collect()] == [(10, 2)]
events = updates.distinct()
assert events.count() == updates.dropDuplicates().count() == 7
events.explain()

10. Duplicato per chiave

Dopo aver eliminato i duplicati esatti, la riga (101, 1) ha ancora tre eventi. Condividono la chiave della riga d’ordine e differiscono in altri campi.

11. Eliminare i duplicati per chiave

dropDuplicates(line_key) conserva una riga per chiave. Usalo quando conta solo la chiave o quando i contenuti dei candidati sono equivalenti.

12. dropDuplicates può conservare un candidato qualsiasi

L’operazione non stabilisce quale revisione sopravvive. Ordinare l’input prima dell’operazione non garantisce quale riga verrà mantenuta.

13. Regola deterministica

Scegli prima la revisione più alta, poi l’ID evento più alto. Dichiara entrambe le scelte e verifica che l’ordinamento distingua i candidati.

valid = events.filter("order_id IS NOT NULL AND line_id IS NOT NULL")
repeated = valid.groupBy(*line_key).count().filter("count > 1")
assert [(r.order_id, r.line_id, r["count"]) for r in repeated.collect()] == [(101, 1, 3)]
assert valid.dropDuplicates(line_key).count() == 4
assert valid.groupBy(*line_key, "revision", "event_id").count().filter("count > 1").count() == 0

14. Valore della riga con il massimo

max_by(value, ordering) restituisce il valore associato al massimo secondo l’ordinamento. Una struct può contenere i campi di ordinamento e un’altra quelli da conservare.

15. Colonna di spareggio

Due eventi condividono la revisione 2. L’ID evento 12 risolve il pari merito. Se rimane una parità, aggiungi un campo stabile che esprima la regola voluta o gestisci il conflitto esplicitamente.

16. max_by o row_number

Usa max_by() per aggregare direttamente a un contenuto per chiave. Usa row_number() per classificare i candidati o ispezionare le righe scartate. Qui entrambe usano lo stesso ordinamento.

17. Eliminare i duplicati

Rimuovi le copie esatte, separa le chiavi mancanti, scegli la riga da conservare e verifica la grana dell’output. Conserva gli eventi scartati se serve la cronologia.

payload = F.struct("revision", "event_id", "units")
ordering = F.struct("revision", "event_id")
max_rows = valid.groupBy(*line_key).agg(F.max_by(payload, ordering).alias("winner"))
max_rows = max_rows.select(*line_key, "winner.*")
window = Window.partitionBy(*line_key).orderBy(F.desc("revision"), F.desc("event_id"))
ranked = valid.withColumn("position", F.row_number().over(window))
latest = ranked.filter("position = 1").drop("position")
assert {tuple(r) for r in max_rows.collect()} == {tuple(r) for r in latest.collect()}
assert latest.filter("order_id = 101 AND line_id = 1").first().event_id == 12
assert latest.groupBy(*line_key).count().filter("count > 1").count() == 0

18. Spark non controlla le chiavi esterne

Un join DataFrame ordinario non rifiuta le chiavi orfane. Scrivi un controllo esplicito di integrità referenziale. L’ultima riga per l’ordine 999 è ancora priva di intestazione.

19. Cardinalità

La cardinalità del join descrive quante righe per lato possono condividere una chiave. Dipende dai dati e dalle colonne del join. Unire un’intestazione univoca alle sue righe dà una relazione uno a molti.

20. Join molti a molti

L’ordine 101 ha due righe aggiornate e due pagamenti. Unendoli per ID ordine, ogni riga si abbina a entrambi i pagamenti.

21. Misurare la cardinalità

Conta le righe per chiave su ciascun lato prima del join. Cerca chiavi ripetute su entrambi i lati.

22. Regola dei conteggi

Un inner join per uguaglianza emette, per ciascuna chiave corrispondente, il prodotto dei conteggi di sinistra e di destra. Somma i prodotti per ottenere il totale. Un left join mantiene anche una riga di output per ogni riga sinistra senza corrispondenza.

23. Controllare due tabelle prima del join

Verifica presenza e unicità attesa delle chiavi, copertura delle corrispondenze e grana dell’output. Gli ID pagamento sono univoci, mentre gli ID ordine nei pagamenti possono ripetersi.

payments = spark.createDataFrame(
    [(1, 101, 400), (2, 101, 600), (3, 102, 500), (4, 103, 300)],
    "payment_id INT, order_id INT, cents INT",
)
assert latest.join(orders, "order_id", "left_anti").count() == 1
assert payments.filter("order_id IS NULL OR payment_id IS NULL").count() == 0
assert payments.select("payment_id").distinct().count() == payments.count()

line_counts = latest.groupBy("order_id").agg(F.count("*").alias("lines"))
payment_counts = payments.groupBy("order_id").agg(F.count("*").alias("payments"))
counts = line_counts.join(payment_counts, "order_id")
assert counts.filter("lines > 1 AND payments > 1").count() == 1
predicted = counts.agg(F.sum(F.col("lines") * F.col("payments"))).first()[0]
joined = latest.join(payments, "order_id")
assert joined.count() == predicted == 5

24. Moltiplicazione delle righe

L’ordine 101 produce quattro righe nel join. Sommando gli importi su queste righe ottieni 2000 centesimi, il doppio dell’importo reale.

25. Aggregare prima del join

Per ottenere una riga per ordine, aggrega separatamente unità e importi alla grana dell’ordine, poi unisci i riepiloghi alle intestazioni.

26. Grana sbagliata

Anche un totale dei pagamenti per ordine si ripete se lo associ a ogni riga dell’ordine. Verifica la grana del risultato prima di sommare le misure.

27. Chiavi senza corrispondenza

Esamina entrambe le direzioni con anti join. Qui una riga aggiornata si riferisce all’ordine 999 e l’intestazione 103 non ha righe. I due casi indicano problemi di copertura diversi.

assert joined.filter("order_id = 101").agg(F.sum("cents")).first()[0] == 2000
units = latest.groupBy("order_id").agg(F.sum("units").alias("units"))
paid = payments.groupBy("order_id").agg(F.sum("cents").alias("paid_cents"))
report = orders.join(units, "order_id", "left").join(paid, "order_id", "left")
assert report.count() == 3
row = report.filter("order_id = 101").first()
assert (row.units, row.paid_cents) == (5, 1000)
assert {r.order_id for r in units.join(orders, "order_id", "left_anti").collect()} == {999}
assert {r.order_id for r in orders.join(units, "order_id", "left_anti").collect()} == {103}
report.orderBy("order_id").show()

Verifica la comprensione

1. Quali campi identificano una riga d’ordine?
2. Quale evento viene mantenuto per la riga (101, 1) e perché?
3. Perché un ordine con due righe e due pagamenti produce quattro righe nel join?

Risposte

1. La coppia order_id e line_id.
2. L’evento 12 ha la revisione più alta e vince lo spareggio sull’ID evento.
3. Ogni riga si abbina a ciascun pagamento dello stesso ordine.