12 · Combinare tabelle

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

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

spark = SparkSession.builder.master("local[2]").appName("12-combine-tables").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
spark.conf.set("spark.sql.ansi.enabled", "true")
schema = "order_id INT, shop_id INT, units INT"
morning = spark.createDataFrame([(101, 1, 2), (102, 2, 3)], schema)
afternoon = spark.createDataFrame([(103, 3, 4), (104, None, 1)], schema)
shops = spark.createDataFrame(
    [(1, "North", "N"), (2, "South", "S"), (4, "West", "W")],
    "shop_id INT, shop STRING, region_code STRING",
)

1. Impilare le righe di due tabelle

Usa union() per accodare lotti con lo stesso ordine delle colonne. Ogni riga di input contribuisce con una riga di output.

2. Unire due tabelle per chiave

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

3. Prodotto cartesiano

crossJoin() abbina ogni ordine a ogni negozio. Quattro ordini e tre negozi producono dodici coppie.

4. Trovare righe senza corrispondenza

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

orders = morning.union(afternoon)
assert orders.count() == morning.count() + afternoon.count() == 4
assert orders.join(shops, "shop_id").count() == 2
assert orders.crossJoin(shops).count() == 12
unmatched = orders.join(shops, "shop_id", "left_anti")
assert {r.order_id for r in unmatched.collect()} == {103, 104}

5. Join sul nome della colonna

Passando un nome comune usi l’uguaglianza e ottieni una sola copia della chiave nell’output.

6. Join su più colonne

Passa una lista di nomi comuni per richiedere che coincidano tutti. Le rettifiche sotto usano la combinazione di ID ordine e ID negozio.

7. Join con un’espressione

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

8. Nome ambiguo dopo un join

Un riferimento senza qualificatore può essere ambiguo se entrambi gli input hanno una colonna shop_id. Assegna alias agli input e seleziona con nomi qualificati.

named = orders.join(shops, "shop_id")
assert named.columns.count("shop_id") == 1
adjustments = spark.createDataFrame(
    [(101, 1, "member"), (101, 2, "other shop")], "order_id INT, shop_id INT, offer STRING"
)
assert orders.join(adjustments, ["order_id", "shop_id"]).first().offer == "member"
o = orders.alias("o")
s = shops.alias("s")
expression_join = o.join(s, F.col("o.shop_id") == F.col("s.shop_id"))
assert expression_join.columns.count("shop_id") == 2
expression_join.select("o.order_id", F.col("o.shop_id").alias("shop_id"), "s.shop").show()

9. Tipi di join

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

10. crossJoin tra tabelle grandi

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

11. Semi join

Un left semi join conserva ogni riga sinistra che ha almeno una corrispondenza. Altri abbinamenti a destra non moltiplicano la riga sinistra.

12. Join senza uguaglianza

Usa disuguaglianze per associare un ordine a una fascia di quantità. I limiti semiaperti includono l’estremo inferiore ed escludono quello superiore.

13. Join senza uguaglianza su tabelle grandi

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

expected = {"inner": 2, "left": 4, "right": 3, "full": 5, "left_semi": 2, "left_anti": 2}
for join_type, count in expected.items():
    assert orders.join(shops, "shop_id", join_type).count() == count

bands = spark.createDataFrame([(0, 3, "small"), (3, 10, "large")], "lo INT, hi INT, band STRING")
banded = orders.join(bands, (orders.units >= bands.lo) & (orders.units < bands.hi))
assert {r.order_id: r.band for r in banded.collect()} == {
    101: "small", 102: "large", 103: "large", 104: "small"
}
banded.explain()

14. Filtro dopo un left join

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

after = orders.join(shops, "shop_id", "left").filter("region_code = 'N'")
before = orders.join(shops.filter("region_code = 'N'"), "shop_id", "left")
assert after.count() == 1
assert before.count() == 4

15. Join con trasmissione

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

16. Join con ordinamento e fusione

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

17. Soglia di trasmissione

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.

18. Quando il suggerimento di broadcast viene ignorato

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

19. Trasmettere una tabella troppo grande

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

keys = ("spark.sql.autoBroadcastJoinThreshold", "spark.sql.adaptive.enabled")
previous = {key: spark.conf.get(key) for key in keys}
try:
    spark.conf.set(keys[0], "-1")
    spark.conf.set(keys[1], "false")
    orders.join(F.broadcast(shops), "shop_id", "left").explain()
    orders.hint("merge").join(shops, "shop_id").explain()
    orders.join(F.broadcast(shops), "shop_id", "full").explain()
finally:
    for key, value in previous.items():
        spark.conf.set(key, value)

20. Catena di join

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

21. Inner join in una catena con orfani

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

22. Join e merge di pandas

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

regions = spark.createDataFrame([("N", "North region")], "region_code STRING, region STRING")
with_shop = orders.join(shops, "shop_id", "left")
assert with_shop.count() == 4
assert with_shop.join(regions, "region_code", "left").count() == 4
assert with_shop.join(regions, "region_code", "inner").count() == 1

23. Unione per nome

unionByName() allinea i nomi corrispondenti. Con allowMissingColumns=True, Spark riempie i campi mancanti con null.

24. union conserva i duplicati

Accodare due volte la stessa riga produce due copie. Usa distinct() solo se i duplicati esatti devono essere accorpati.

25. union allinea per posizione

Riordinare i campi può mettere silenziosamente valori sotto nomi sbagliati quando i tipi restano compatibili.

26. Unione con tipi diversi

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

reordered = afternoon.select("units", "shop_id", "order_id")
assert {r.order_id for r in morning.unionByName(reordered).collect()} == {101, 102, 103, 104}
assert {r.order_id for r in morning.union(reordered).collect()} == {101, 102, 4, 1}
assert morning.union(morning).count() == 4
assert morning.union(morning).distinct().count() == 2

missing_units = morning.unionByName(afternoon.drop("units"), allowMissingColumns=True)
assert missing_units.filter("units IS NULL").count() == 2
widened = morning.unionByName(afternoon.withColumn("units", F.col("units").cast("bigint")))
assert widened.schema["units"].dataType.simpleString() == "bigint"

27. Intersezione

intersect() restituisce righe condivise distinte. intersectAll() conserva ciascuna riga condivisa fino alla minore molteplicità tra i due input.

28. Differenza

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.

29. subtract elimina i duplicati

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.

30. Confrontare due versioni di una tabella

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

old = spark.createDataFrame([(1, 2), (1, 2), (2, 3), (3, None)], "order_id INT, units INT")
new = spark.createDataFrame([(1, 2), (2, 4), (3, None), (4, 1)], old.schema)
assert old.intersect(new).count() == 2
assert old.intersectAll(old).count() == 4
assert {tuple(r) for r in old.subtract(new).collect()} == {(2, 3)}
removed = old.exceptAll(new)
added = new.exceptAll(old)
assert sorted(tuple(r) for r in removed.collect()) == [(1, 2), (2, 3)]
assert sorted(tuple(r) for r in added.collect()) == [(2, 4), (4, 1)]
removed.orderBy("order_id").show()
added.orderBy("order_id").show()

Verifica la comprensione

1. Quale join trova ordini senza corrispondenza?
2. Come mantieni tutti gli ordini limitando le righe ammissibili della tabella di riferimento?
3. Quale operazione rileva un cambio nella molteplicità tra snapshot?

Risposte

1. Un left anti join.
2. Filtra la tabella di riferimento prima del left join o inserisci la condizione nell’espressione del join.
3. Usa exceptAll() in entrambe le direzioni su colonne allineate.