{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 12 · Combinare tabelle\n\nCombina righe per chiave o condizione, accoda lotti e confronta snapshot tenendo conto dei duplicati. Usa PySpark classico 4.0.x e Java 17+. Su Kaggle, se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"` in una cella di preparazione, poi usa Run All. Tutti i dati sono nel notebook.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"12-combine-tables\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")\nschema = \"order_id INT, shop_id INT, units INT\"\nmorning = spark.createDataFrame([(101, 1, 2), (102, 2, 3)], schema)\nafternoon = spark.createDataFrame([(103, 3, 4), (104, None, 1)], schema)\nshops = spark.createDataFrame(\n    [(1, \"North\", \"N\"), (2, \"South\", \"S\"), (4, \"West\", \"W\")],\n    \"shop_id INT, shop STRING, region_code STRING\",\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Impilare le righe di due tabelle\n\nUsa `union()` per accodare lotti con lo stesso ordine delle colonne. Ogni riga di input contribuisce con una riga di output.\n\n## 2. Unire due tabelle per chiave\n\nUn join abbina righe dei due input. Un inner join conserva gli abbinamenti. Qui ogni ordine corrisponde a zero o un negozio perché la chiave della tabella di riferimento è univoca. Se a una chiave corrispondono più righe, un left join può emetterne più di una per ordine.\n\n## 3. Prodotto cartesiano\n\n`crossJoin()` abbina ogni ordine a ogni negozio. Quattro ordini e tre negozi producono dodici coppie.\n\n## 4. Trovare righe senza corrispondenza\n\nUn left anti join conserva le righe a sinistra senza corrispondenza a destra. Una chiave presente ma assente dalla tabella di riferimento è orfana. Con l’uguaglianza ordinaria anche una chiave null resta senza corrispondenza, quindi l’anti join può includere entrambi i casi.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "orders = morning.union(afternoon)\nassert orders.count() == morning.count() + afternoon.count() == 4\nassert orders.join(shops, \"shop_id\").count() == 2\nassert orders.crossJoin(shops).count() == 12\nunmatched = orders.join(shops, \"shop_id\", \"left_anti\")\nassert {r.order_id for r in unmatched.collect()} == {103, 104}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 5. Join sul nome della colonna\n\nPassando un nome comune usi l’uguaglianza e ottieni una sola copia della chiave nell’output.\n\n## 6. Join su più colonne\n\nPassa una lista di nomi comuni per richiedere che coincidano tutti. Le rettifiche sotto usano la combinazione di ID ordine e ID negozio.\n\n## 7. Join con un’espressione\n\nUsa un’espressione Column se le chiavi hanno nomi diversi o serve un’altra condizione. Questa forma può conservare le colonne di entrambi gli input, incluse le due copie di una chiave.\n\n## 8. Nome ambiguo dopo un join\n\nUn riferimento senza qualificatore può essere ambiguo se entrambi gli input hanno una colonna `shop_id`. Assegna alias agli input e seleziona con nomi qualificati.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "named = orders.join(shops, \"shop_id\")\nassert named.columns.count(\"shop_id\") == 1\nadjustments = spark.createDataFrame(\n    [(101, 1, \"member\"), (101, 2, \"other shop\")], \"order_id INT, shop_id INT, offer STRING\"\n)\nassert orders.join(adjustments, [\"order_id\", \"shop_id\"]).first().offer == \"member\"\no = orders.alias(\"o\")\ns = shops.alias(\"s\")\nexpression_join = o.join(s, F.col(\"o.shop_id\") == F.col(\"s.shop_id\"))\nassert expression_join.columns.count(\"shop_id\") == 2\nexpression_join.select(\"o.order_id\", F.col(\"o.shop_id\").alias(\"shop_id\"), \"s.shop\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. Tipi di join\n\nInner conserva le coppie corrispondenti. Left e right outer preservano i rispettivi lati. Full outer mantiene le righe senza corrispondenza da entrambi. Semi e anti restituiscono solo le colonne di sinistra. I conteggi valgono per questi dati con chiave di riferimento univoca.\n\n## 10. crossJoin tra tabelle grandi\n\nLa dimensione dell’output è il prodotto dei conteggi. Due tabelle con un milione di righe ciascuna possono produrre mille miliardi di coppie. Qui si esegue solo il piccolo esempio.\n\n## 11. Semi join\n\nUn left semi join conserva ogni riga sinistra che ha almeno una corrispondenza. Altri abbinamenti a destra non moltiplicano la riga sinistra.\n\n## 12. Join senza uguaglianza\n\nUsa disuguaglianze per associare un ordine a una fascia di quantità. I limiti semiaperti includono l’estremo inferiore ed escludono quello superiore.\n\n## 13. Join senza uguaglianza su tabelle grandi\n\nSenza una chiave di uguaglianza, Spark può dover confrontare molti candidati. Osserva il piano e conta gli abbinamenti su un piccolo campione. Dove i dati lo permettono, aggiungi una chiave di uguaglianza che preservi gli abbinamenti validi e restringa i candidati.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "expected = {\"inner\": 2, \"left\": 4, \"right\": 3, \"full\": 5, \"left_semi\": 2, \"left_anti\": 2}\nfor join_type, count in expected.items():\n    assert orders.join(shops, \"shop_id\", join_type).count() == count\n\nbands = spark.createDataFrame([(0, 3, \"small\"), (3, 10, \"large\")], \"lo INT, hi INT, band STRING\")\nbanded = orders.join(bands, (orders.units >= bands.lo) & (orders.units < bands.hi))\nassert {r.order_id: r.band for r in banded.collect()} == {\n    101: \"small\", 102: \"large\", 103: \"large\", 104: \"small\"\n}\nbanded.explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Filtro dopo un left join\n\nUn filtro successivo su un campo di destra può eliminare righe senza corrispondenza perché quel campo è null. Se la regola definisce gli abbinamenti ammissibili e vuoi mantenere tutti gli ordini, filtra la tabella di riferimento prima del join.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "after = orders.join(shops, \"shop_id\", \"left\").filter(\"region_code = 'N'\")\nbefore = orders.join(shops.filter(\"region_code = 'N'\"), \"shop_id\", \"left\")\nassert after.count() == 1\nassert before.count() == 4",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Join con trasmissione\n\nUn broadcast join rende il lato da trasmettere disponibile ai worker. L’altro lato può così essere elaborato senza rimescolarlo per chiave. Usa `F.broadcast()` per indicare che la tabella di riferimento è abbastanza piccola.\n\n## 16. Join con ordinamento e fusione\n\nUn sort-merge join riunisce le chiavi corrispondenti e fonde gli input ordinati. L’esempio richiede la strategia di fusione su un join per uguaglianza e stampa il piano fisico.\n\n## 17. Soglia di trasmissione\n\n`spark.sql.autoBroadcastJoinThreshold` controlla la trasmissione automatica in base alla dimensione nel piano iniziale. L’esempio disattiva temporaneamente broadcast automatico ed esecuzione adattiva per rendere il confronto più leggibile.\n\n## 18. Quando il suggerimento di broadcast viene ignorato\n\nIl suggerimento può superare la soglia automatica, ma strategia e lato trasmesso devono essere compatibili con il tipo di join. Un full outer per uguaglianza non può usare un broadcast hash join. Esamina la strategia selezionata nel piano.\n\n## 19. Trasmettere una tabella troppo grande\n\nLa trasmissione richiede preparazione e distribuzione dei dati. Un lato troppo grande può causare pressione sulla memoria o timeout. Usa suggerimenti per tabelle di riferimento di dimensione nota.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "keys = (\"spark.sql.autoBroadcastJoinThreshold\", \"spark.sql.adaptive.enabled\")\nprevious = {key: spark.conf.get(key) for key in keys}\ntry:\n    spark.conf.set(keys[0], \"-1\")\n    spark.conf.set(keys[1], \"false\")\n    orders.join(F.broadcast(shops), \"shop_id\", \"left\").explain()\n    orders.hint(\"merge\").join(shops, \"shop_id\").explain()\n    orders.join(F.broadcast(shops), \"shop_id\", \"full\").explain()\nfinally:\n    for key, value in previous.items():\n        spark.conf.set(key, value)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Catena di join\n\nArricchisci gli ordini prima con il negozio e poi con la regione. Controlla il numero di righe a ogni passaggio. Chiavi di riferimento univoche danno la relazione molti a uno desiderata.\n\n## 21. Inner join in una catena con orfani\n\nUn left join conserva gli ordini senza negozio. Il successivo inner join con le regioni elimina le righe senza regione corrispondente, inclusi gli ordini già privi di negozio.\n\n## 22. Join e merge di pandas\n\nUn join Spark per chiave è simile a `merge(..., on=..., how=...)` di pandas per le combinazioni inner e outer di base. pandas può abbinare chiavi mancanti tra loro. L’uguaglianza ordinaria di Spark non abbina null.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "regions = spark.createDataFrame([(\"N\", \"North region\")], \"region_code STRING, region STRING\")\nwith_shop = orders.join(shops, \"shop_id\", \"left\")\nassert with_shop.count() == 4\nassert with_shop.join(regions, \"region_code\", \"left\").count() == 4\nassert with_shop.join(regions, \"region_code\", \"inner\").count() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 23. Unione per nome\n\n`unionByName()` allinea i nomi corrispondenti. Con `allowMissingColumns=True`, Spark riempie i campi mancanti con null.\n\n## 24. union conserva i duplicati\n\nAccodare due volte la stessa riga produce due copie. Usa `distinct()` solo se i duplicati esatti devono essere accorpati.\n\n## 25. union allinea per posizione\n\nRiordinare i campi può mettere silenziosamente valori sotto nomi sbagliati quando i tipi restano compatibili.\n\n## 26. Unione con tipi diversi\n\nSpark può promuovere tipi compatibili come INT e BIGINT a un tipo comune. L’allineamento per nome è soggetto alle regole dei tipi. Converti esplicitamente quando il campo ha un tipo di destinazione definito.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "reordered = afternoon.select(\"units\", \"shop_id\", \"order_id\")\nassert {r.order_id for r in morning.unionByName(reordered).collect()} == {101, 102, 103, 104}\nassert {r.order_id for r in morning.union(reordered).collect()} == {101, 102, 4, 1}\nassert morning.union(morning).count() == 4\nassert morning.union(morning).distinct().count() == 2\n\nmissing_units = morning.unionByName(afternoon.drop(\"units\"), allowMissingColumns=True)\nassert missing_units.filter(\"units IS NULL\").count() == 2\nwidened = morning.unionByName(afternoon.withColumn(\"units\", F.col(\"units\").cast(\"bigint\")))\nassert widened.schema[\"units\"].dataType.simpleString() == \"bigint\"",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 27. Intersezione\n\n`intersect()` restituisce righe condivise distinte. `intersectAll()` conserva ciascuna riga condivisa fino alla minore molteplicità tra i due input.\n\n## 28. Differenza\n\n`exceptAll()` sottrae le occorrenze di destra da quelle di sinistra. Confronta colonne nello stesso ordine. Le operazioni insiemistiche confrontano righe intere e considerano uguali i null in quel confronto.\n\n## 29. subtract elimina i duplicati\n\n`subtract()` restituisce le righe distinte di sinistra assenti a destra. Se una riga compare a destra anche una sola volta, nessuna copia di sinistra sopravvive.\n\n## 30. Confrontare due versioni di una tabella\n\nAllinea prima nomi, ordine e tipi delle colonne. Usa `exceptAll()` in entrambe le direzioni per individuare occorrenze rimosse e aggiunte. Una riga modificata appare come rimozione e aggiunta. Per accoppiare gli aggiornamenti per chiave di business, verificane prima l’unicità in entrambe le versioni.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "old = spark.createDataFrame([(1, 2), (1, 2), (2, 3), (3, None)], \"order_id INT, units INT\")\nnew = spark.createDataFrame([(1, 2), (2, 4), (3, None), (4, 1)], old.schema)\nassert old.intersect(new).count() == 2\nassert old.intersectAll(old).count() == 4\nassert {tuple(r) for r in old.subtract(new).collect()} == {(2, 3)}\nremoved = old.exceptAll(new)\nadded = new.exceptAll(old)\nassert sorted(tuple(r) for r in removed.collect()) == [(1, 2), (2, 3)]\nassert sorted(tuple(r) for r in added.collect()) == [(2, 4), (4, 1)]\nremoved.orderBy(\"order_id\").show()\nadded.orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale join trova ordini senza corrispondenza?\n2. Come mantieni tutti gli ordini limitando le righe ammissibili della tabella di riferimento?\n3. Quale operazione rileva un cambio nella molteplicità tra snapshot?\n\n### Risposte\n\n1. Un left anti join.\n2. Filtra la tabella di riferimento prima del left join o inserisci la condizione nell’espressione del join.\n3. Usa `exceptAll()` in entrambe le direzioni su colonne allineate.",
      "id": "cell-19"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}