09 · Usare tipi complessi

Ispeziona e trasforma dati d’ordine annidati con array, mappe e struct. Infine interpreta JSON come struct tipizzata o VARIANT. Esegui tutte le celle in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa PySpark in una cella di preparazione con %pip install "pyspark>=4.0,<4.1". Tutti i dati sono definiti nel notebook.

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

spark = SparkSession.builder.master("local[2]").appName("09-use-complex-types").getOrCreate()
spark.conf.set("spark.sql.ansi.enabled", "true")
orders = spark.createDataFrame(
    [
        (101, ["tea", "cake", "tea"], {"tea": 2, "cake": 1}, ("counter", 4)),
        (102, [], {}, ("takeaway", 0)),
        (103, None, None, None),
    ],
    "order_id INT, items ARRAY<STRING>, quantities MAP<STRING,INT>, pickup STRUCT<channel:STRING, wait_minutes:INT>",
)
orders.printSchema()

1. Array

Un array è una raccolta ordinata di elementi dello stesso tipo. Può contenere duplicati, come nell’ordine 101.

2. Separare un array in righe

explode() produce una riga per elemento. Gli array null e vuoti non producono righe.

3. Dimensione di un array

array_size() restituisce il numero di elementi, zero per un array vuoto e null per un array null.

4. Leggere un elemento di un array

get() legge l’elemento con indice a partire da zero e restituisce null per un indice fuori intervallo.

5. Cercare un valore

array_contains() verifica l’appartenenza. Un array null dà null.

items = F.col("items")
exploded = orders.select("order_id", F.explode(items).alias("item"))
assert exploded.count() == 3
exploded.show()
orders.select(
    "order_id", F.array_size(items).alias("size"),
    F.get(items, 0).alias("first_item"), F.array_contains(items, "tea").alias("has_tea"),
).orderBy("order_id").show()

6. Operazioni su insiemi tra array

array_union(), array_intersect() e array_except() restituiscono gli elementi distinti dell’unione, intersezione e differenza. Tratta i risultati come insiemi quando l’ordine è irrilevante.

7. element_at conta da 1 e [i] da 0

Per gli array, element_at(items, 1) e items[0] leggono entrambi il primo elemento. Un indice negativo in element_at conta dalla fine. In modalità ANSI una posizione non valida può generare un errore. try_element_at() restituisce null per posizioni fuori intervallo, mentre l’indice zero resta non valido.

8. Unire gli elementi in un testo

array_join() costruisce una stringa con un separatore. Salta gli elementi null a meno che tu non fornisca un sostituto.

9. Funzioni di ordine superiore

transform() applica un’espressione a ogni elemento, filter() seleziona gli elementi e aggregate() li combina in un valore. Le lambda Python costruiscono espressioni di colonna Spark.

one_order = orders.filter("order_id = 101")
menu = F.array(F.lit("tea"), F.lit("coffee"))
one_order.select(
    F.array_union(items, menu).alias("union"),
    F.array_intersect(items, menu).alias("intersection"),
    F.array_except(items, menu).alias("difference"),
).show()
indexes = one_order.select(F.element_at(items, 1), items[0], F.element_at(items, -1)).first()
assert tuple(indexes) == ("tea", "tea", "tea")

one_order.select(F.array_join(items, ", ").alias("label")).show()
orders.select(
    "order_id",
    F.transform(items, lambda item: F.upper(item)).alias("upper_items"),
    F.filter(items, lambda item: item == "tea").alias("teas"),
    F.aggregate(items, F.lit(0), lambda count, item: count + F.length(item)).alias("characters"),
).show()

10. Mappa

Una mappa memorizza valori indicizzati da chiavi univoche e non null. La mappa delle quantità indica il numero di unità ordinate per prodotto.

11. Creare una mappa

create_map() accetta espressioni alternate di chiave e valore. Racchiudi le chiavi e i valori letterali in lit().

12. Chiavi e valori di una mappa

map_keys() e map_values() restituiscono array senza ordine garantito. Usa map_entries() per mantenere ogni chiave associata al suo valore.

13. Leggere il valore di una chiave

element_at(map, key) restituisce il valore corrispondente o null se la chiave manca.

14. Separare una mappa in righe

explode() emette una chiave e un valore per ogni voce.

15. Valori di una mappa dello stesso tipo

Una mappa Spark ha un tipo dichiarato per le chiavi e uno per i valori. Usa una struct se servono campi con nome e tipi diversi.

quantities = F.col("quantities")
defaults = F.create_map(F.lit("tea"), F.lit(0), F.lit("cake"), F.lit(0))
orders.select("order_id", defaults.alias("defaults"), F.map_keys(quantities), F.map_values(quantities)).show()
assert tuple(one_order.select(F.element_at(quantities, "tea"), F.element_at(quantities, "coffee")).first()) == (2, None)
entries = orders.select("order_id", F.explode(quantities).alias("item", "units"))
assert {(r.item, r.units) for r in entries.collect()} == {("tea", 2), ("cake", 1)}

16. Struttura (struct)

Una struct raggruppa campi con nome, ognuno con il proprio tipo. Qui il ritiro contiene un canale testuale e un tempo di attesa intero. struct() costruisce una struttura da espressioni.

17. Leggere un campo di una struttura

Usa percorsi con punti in select() oppure getField() su una colonna struct.

18. Portare i campi in colonne separate

pickup.* espande i campi immediati della struttura. Il risultato mantiene una riga per ordine di input.

19. Aggiungere o togliere campi di una struttura

withField() aggiunge o sostituisce un campo. dropFields() rimuove i campi indicati. Una struct genitore null rimane null.

orders.select("order_id", F.struct("order_id", "pickup").alias("order_details")).show(truncate=False)
assert one_order.select(F.col("pickup").getField("channel")).first()[0] == "counter"
expanded = orders.select("order_id", "pickup.*")
assert expanded.columns == ["order_id", "channel", "wait_minutes"]
edited = orders.select(
    "order_id",
    F.col("pickup").withField("confirmed", F.lit(True)).dropFields("wait_minutes").alias("pickup"),
)
edited.printSchema()
edited.show()

20. Tenere array vuoti e null

explode_outer() emette una riga segnaposto con elemento null per ogni array vuoto o null. Conserva la dimensione originale per distinguere i due casi.

21. Posizione dell’elemento

posexplode_outer() aggiunge una posizione a partire da zero. Le righe segnaposto hanno posizione null.

22. explode cambia la grana

L’input ha una riga per ordine. L’output espanso ha una riga per elemento, più le righe segnaposto con la variante outer. Contare le righe espanse dà un risultato diverso dal numero di ordini.

outer = orders.select("order_id", F.array_size(items).alias("original_size"),
                      F.explode_outer(items).alias("item"))
assert outer.count() == 5
positioned = orders.select("order_id", F.posexplode_outer(items).alias("position", "item"))
positioned.orderBy("order_id", "position").show()
assert [r.position for r in positioned.filter("order_id = 101").orderBy("position").collect()] == [0, 1, 2]

23. Leggere testo JSON in una colonna

from_json() interpreta il testo secondo uno schema. Il risultato qui sotto è una struct che Spark può ispezionare campo per campo.

24. Scrivere una struttura come JSON

to_json() serializza la struttura in una stringa. Serializza i campi interpretati, che possono differire dal contenuto originale.

25. Estrarre un campo da testo JSON

get_json_object() legge un percorso dal testo originale. La colonna restituita è una stringa.

26. VARIANT

Spark 4 aggiunge VARIANT per dati semistrutturati con forma variabile. try_parse_json() restituisce null SQL per JSON malformato e try_variant_get() legge un percorso con il tipo richiesto, restituendo null se il percorso manca o la conversione fallisce.

27. from_json con schema incompatibile

In modalità permissiva, un oggetto JSON valido può dare campi null se i nomi o i tipi differiscono da quelli dichiarati. Conserva l’input originale e verifica i campi obbligatori. Un campo null da solo non distingue uno schema incompatibile da un campo assente o da un null JSON esplicito.

payloads = spark.createDataFrame(
    [(1, '{"channel":"counter","wait_minutes":4}'),
     (2, '{"channel":"takeaway","wait_minutes":"unknown"}'),
     (3, 'not JSON')],
    "payload_id INT, raw_json STRING",
)
schema = "channel STRING, wait_minutes INT"
parsed = payloads.withColumn("pickup", F.from_json("raw_json", schema, {"mode": "PERMISSIVE"}))
parsed.select("payload_id", "pickup.*", F.to_json("pickup").alias("serialized"),
              F.get_json_object("raw_json", "$.wait_minutes").alias("wait_text")).show(truncate=False)

variants = payloads.withColumn("payload", F.try_parse_json("raw_json"))
wait = variants.select("payload_id", F.try_variant_get("payload", "$.wait_minutes", "int").alias("wait"))
assert {r.payload_id: r.wait for r in wait.collect()} == {1: 4, 2: None, 3: None}

waits = parsed.select("payload_id", F.col("pickup.wait_minutes").alias("wait")).collect()
assert {r.payload_id: r.wait for r in waits} == {1: 4, 2: None, 3: None}
wrong_schema = payloads.filter("payload_id = 1").select(
    F.from_json("raw_json", "wait_seconds INT").getField("wait_seconds").alias("wait_seconds")
)
assert wrong_schema.first().wait_seconds is None

Verifica la comprensione

1. Come conservi un ordine con array vuoto quando espandi i prodotti?
2. Quale tipo annidato permette al canale e al tempo di attesa di avere tipi diversi?
3. Perché conservare il JSON originale quando verifichi una conversione tipizzata?

Risposte

1. Usa explode_outer() e conserva la dimensione originale se devi distinguere array vuoti e null.
2. Usa una struct con campi distinti e dotati di nome.
3. Un null tipizzato può avere più cause. Il contenuto originale consente di indagarle.