{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 13 · Controllare chiavi e duplicati\n\nVerifica 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.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession, Window\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"13-check-keys\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nupdates = spark.createDataFrame(\n    [(101, 1, 1, 10, 2), (101, 1, 1, 10, 2), (101, 1, 2, 11, 3),\n     (101, 1, 2, 12, 4), (101, 2, 1, 13, 1), (102, 1, 1, 14, 2),\n     (999, 1, 1, 15, 5), (None, 1, 1, 16, 1)],\n    \"order_id INT, line_id INT, revision INT, event_id INT, units INT\",\n)\norders = spark.createDataFrame([(101,), (102,), (103,)], \"order_id INT\")\nline_key = [\"order_id\", \"line_id\"]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Duplicato esatto\n\nDue righe sono duplicati esatti quando ogni colonna ha lo stesso valore. Le due copie dell’evento 10 coincidono in tutti i campi.\n\n## 2. Chiave primaria\n\nUna chiave primaria identifica una riga e deve essere univoca e presente. La tabella delle intestazioni usa `order_id`. Verifica entrambe le proprietà nel DataFrame.\n\n## 3. Chiave composta\n\nUna 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.\n\n## 4. Chiave esterna\n\nL’`order_id` della riga fa riferimento alla tabella delle intestazioni. I valori possono ripetersi perché un ordine può avere più righe.\n\n## 5. Orfano\n\nL’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.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert orders.filter(\"order_id IS NULL\").count() == 0\nassert orders.select(\"order_id\").distinct().count() == orders.count()\norphans = updates.filter(\"order_id IS NOT NULL\").join(orders, \"order_id\", \"left_anti\")\nassert {r.order_id for r in orphans.collect()} == {999}\nassert updates.filter(\"order_id IS NULL\").count() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Eliminare righe duplicate\n\nDecidi quali righe considerare duplicate prima di eliminarle. Acquisizioni ripetute di uno stesso evento e revisioni diverse di una riga d’ordine richiedono regole diverse.\n\n## 7. Eliminare i duplicati esatti\n\n`distinct()` e `dropDuplicates()` senza un sottoinsieme mantengono una copia di ogni riga completa.\n\n## 8. distinct come trasformazione larga\n\nLe copie possono trovarsi in partizioni diverse. Spark in genere usa uno shuffle per riunire le righe uguali. Cerca uno scambio nel piano fisico.\n\n## 9. Trovare i duplicati\n\nRaggruppa per le colonne che definiscono l’identità, conta le righe e conserva i gruppi con più di una riga.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "exact_groups = updates.groupBy(*updates.columns).count().filter(\"count > 1\")\nassert [(r.event_id, r[\"count\"]) for r in exact_groups.collect()] == [(10, 2)]\nevents = updates.distinct()\nassert events.count() == updates.dropDuplicates().count() == 7\nevents.explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Duplicato per chiave\n\nDopo 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.\n\n## 11. Eliminare i duplicati per chiave\n\n`dropDuplicates(line_key)` conserva una riga per chiave. Usalo quando conta solo la chiave o quando i contenuti dei candidati sono equivalenti.\n\n## 12. dropDuplicates può conservare un candidato qualsiasi\n\nL’operazione non stabilisce quale revisione sopravvive. Ordinare l’input prima dell’operazione non garantisce quale riga verrà mantenuta.\n\n## 13. Regola deterministica\n\nScegli prima la revisione più alta, poi l’ID evento più alto. Dichiara entrambe le scelte e verifica che l’ordinamento distingua i candidati.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "valid = events.filter(\"order_id IS NOT NULL AND line_id IS NOT NULL\")\nrepeated = valid.groupBy(*line_key).count().filter(\"count > 1\")\nassert [(r.order_id, r.line_id, r[\"count\"]) for r in repeated.collect()] == [(101, 1, 3)]\nassert valid.dropDuplicates(line_key).count() == 4\nassert valid.groupBy(*line_key, \"revision\", \"event_id\").count().filter(\"count > 1\").count() == 0",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Valore della riga con il massimo\n\n`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.\n\n## 15. Colonna di spareggio\n\nDue 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.\n\n## 16. max_by o row_number\n\nUsa `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.\n\n## 17. Eliminare i duplicati\n\nRimuovi 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.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "payload = F.struct(\"revision\", \"event_id\", \"units\")\nordering = F.struct(\"revision\", \"event_id\")\nmax_rows = valid.groupBy(*line_key).agg(F.max_by(payload, ordering).alias(\"winner\"))\nmax_rows = max_rows.select(*line_key, \"winner.*\")\nwindow = Window.partitionBy(*line_key).orderBy(F.desc(\"revision\"), F.desc(\"event_id\"))\nranked = valid.withColumn(\"position\", F.row_number().over(window))\nlatest = ranked.filter(\"position = 1\").drop(\"position\")\nassert {tuple(r) for r in max_rows.collect()} == {tuple(r) for r in latest.collect()}\nassert latest.filter(\"order_id = 101 AND line_id = 1\").first().event_id == 12\nassert latest.groupBy(*line_key).count().filter(\"count > 1\").count() == 0",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 18. Spark non controlla le chiavi esterne\n\nUn 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.\n\n## 19. Cardinalità\n\nLa 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.\n\n## 20. Join molti a molti\n\nL’ordine 101 ha due righe aggiornate e due pagamenti. Unendoli per ID ordine, ogni riga si abbina a entrambi i pagamenti.\n\n## 21. Misurare la cardinalità\n\nConta le righe per chiave su ciascun lato prima del join. Cerca chiavi ripetute su entrambi i lati.\n\n## 22. Regola dei conteggi\n\nUn 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.\n\n## 23. Controllare due tabelle prima del join\n\nVerifica 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.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "payments = spark.createDataFrame(\n    [(1, 101, 400), (2, 101, 600), (3, 102, 500), (4, 103, 300)],\n    \"payment_id INT, order_id INT, cents INT\",\n)\nassert latest.join(orders, \"order_id\", \"left_anti\").count() == 1\nassert payments.filter(\"order_id IS NULL OR payment_id IS NULL\").count() == 0\nassert payments.select(\"payment_id\").distinct().count() == payments.count()\n\nline_counts = latest.groupBy(\"order_id\").agg(F.count(\"*\").alias(\"lines\"))\npayment_counts = payments.groupBy(\"order_id\").agg(F.count(\"*\").alias(\"payments\"))\ncounts = line_counts.join(payment_counts, \"order_id\")\nassert counts.filter(\"lines > 1 AND payments > 1\").count() == 1\npredicted = counts.agg(F.sum(F.col(\"lines\") * F.col(\"payments\"))).first()[0]\njoined = latest.join(payments, \"order_id\")\nassert joined.count() == predicted == 5",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 24. Moltiplicazione delle righe\n\nL’ordine 101 produce quattro righe nel join. Sommando gli importi su queste righe ottieni 2000 centesimi, il doppio dell’importo reale.\n\n## 25. Aggregare prima del join\n\nPer ottenere una riga per ordine, aggrega separatamente unità e importi alla grana dell’ordine, poi unisci i riepiloghi alle intestazioni.\n\n## 26. Grana sbagliata\n\nAnche 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.\n\n## 27. Chiavi senza corrispondenza\n\nEsamina 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.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert joined.filter(\"order_id = 101\").agg(F.sum(\"cents\")).first()[0] == 2000\nunits = latest.groupBy(\"order_id\").agg(F.sum(\"units\").alias(\"units\"))\npaid = payments.groupBy(\"order_id\").agg(F.sum(\"cents\").alias(\"paid_cents\"))\nreport = orders.join(units, \"order_id\", \"left\").join(paid, \"order_id\", \"left\")\nassert report.count() == 3\nrow = report.filter(\"order_id = 101\").first()\nassert (row.units, row.paid_cents) == (5, 1000)\nassert {r.order_id for r in units.join(orders, \"order_id\", \"left_anti\").collect()} == {999}\nassert {r.order_id for r in orders.join(units, \"order_id\", \"left_anti\").collect()} == {103}\nreport.orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quali campi identificano una riga d’ordine?\n2. Quale evento viene mantenuto per la riga (101, 1) e perché?\n3. Perché un ordine con due righe e due pagamenti produce quattro righe nel join?\n\n### Risposte\n\n1. La coppia `order_id` e `line_id`.\n2. L’evento 12 ha la revisione più alta e vince lo spareggio sull’ID evento.\n3. Ogni riga si abbina a ciascun pagamento dello stesso ordine.",
      "id": "cell-15"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}