02 · Creare una tabella

Crea un DataFrame Spark da dati locali, scegli lo schema e converti testo irregolare in valori tipizzati. La lezione è autonoma e usa PySpark classico 4.0.x, Python 3.9+, Java 17+ e pandas 2+. Su Kaggle, se serve, esegui %pip install "pyspark>=4.0,<4.1" "pandas>=2" scikit-learn in una cella di preparazione con Internet abilitato. Poi usa Run All. I dati Iris sono inclusi in scikit-learn.

1. Importare i moduli di PySpark

Importa esplicitamente i nomi usati nella lezione. col() fa riferimento a una colonna per nome. lit() fornisce un valore costante. Entrambe compaiono nell’ultimo esempio.

2. SparkSession

Una SparkSession è il punto di accesso per creare DataFrame e impostare le opzioni SQL.

3. Import con asterisco

Un import come from pyspark.sql.functions import * introduce molti nomi nello spazio dei nomi corrente. Può nascondere nomi Python come sum. Gli import espliciti ne rendono visibile la provenienza.

4. Creare o riusare una sessione

getOrCreate() riusa una sessione disponibile o ne avvia una. Qui, se si parte da zero, local[2] richiede un’applicazione Spark locale con due thread di lavoro.

from datetime import date, datetime
from decimal import Decimal

import pandas as pd
from pyspark.errors import PySparkValueError, PySparkException
from pyspark.sql import Row, SparkSession
from pyspark.sql.functions import col, lit
from pyspark.sql.types import DecimalType, IntegerType, StructField, StructType
from sklearn.datasets import load_iris

spark = SparkSession.builder.master("local[2]").appName("02-create-a-table").getOrCreate()
assert spark.version.startswith("4."), "This lesson requires PySpark 4.x."

5. Caricare una tabella pubblica

Carica Iris in pandas e passa la piccola tabella a Spark. columns elenca i nomi. select() mantiene le colonne richieste, alias() assegna un nome a una colonna di output e distinct() conserva una copia di ogni riga distinta. count() restituisce il numero di righe.

iris_pd = load_iris(as_frame=True).frame
iris = spark.createDataFrame(iris_pd)
print(iris.columns)
labels = iris.select(iris["target"].alias("species_id")).distinct()
assert labels.count() == 3
labels.show()

6. Tipo di dato di una colonna

Una colonna ha un tipo Spark che determina quali valori può contenere e quali operazioni si possono applicare. printSchema() mostra i nomi dei campi, i tipi e la possibilità di avere valori mancanti.

iris.printSchema()

7. Riga come oggetto Row

Row permette di assegnare un nome ai valori di un record. Qui ogni record contiene una riga d’ordine.

8. Creare una tabella da dati Python

Passa a createDataFrame() una lista di oggetti Row. Spark può dedurre i tipi delle colonne dai valori Python.

9. Creare una sequenza di numeri

spark.range(3) restituisce un DataFrame con gli interi lunghi 0, 1 e 2 in una colonna chiamata id.

orders = [Row(order_id=101, drink_name="tea"), Row(order_id=102, drink_name="coffee")]
orders_df = spark.createDataFrame(orders)
orders_df.show()
spark.range(3).show()

10. Schema

Lo schema descrive i campi, il loro ordine, i tipi e la possibilità di contenere null.

11. Nomi delle colonne in snake case

Usa nomi minuscoli con trattini bassi, come order_id. toDF() sostituisce tutti i nomi delle colonne per posizione. Mantieni i nomi nello stesso ordine delle colonne di partenza.

orders_df.printSchema()
renamed_orders = orders_df.toDF("receipt_id", "drink_name")
assert renamed_orders.columns == ["receipt_id", "drink_name"]

12. Avviare un programma PySpark

In un notebook le celle vengono eseguite nello stesso kernel Python. In uno script autonomo, inserisci la creazione della sessione e le operazioni sotto if __name__ == "__main__": e arresta la sessione al termine. Il driver coordina l’applicazione Spark. Qui il kernel del notebook esegue la parte Python del driver.

13. Scrivere lo schema come testo

Una stringa DDL specifica per Spark nomi e tipi delle colonne. In questo esempio fissa anche l’ordine dei campi.

14. None di Python diventa null in Spark

Un valore mancante fornito come None diventa NULL in SQL se il campo ammette null.

quantity_rows = [(101, 2), (102, None)]
quantities = spark.createDataFrame(quantity_rows, "order_id INT, quantity INT")
quantities.show()
quantities.printSchema()

15. Creare una tabella da pandas

Passa un DataFrame pandas a createDataFrame(). Con Iris hai già visto questo passaggio su dati reali. Qui usiamo una piccola tabella pandas con uno schema esplicito.

16. Convertire una tabella in pandas

toPandas() copia il risultato Spark in un DataFrame pandas nel processo del driver.

17. toPandas su dati grandi

L’intero risultato deve entrare nella memoria del driver. Usa il metodo per risultati piccoli, come questa tabella di due righe.

18. Il null di Spark diventa NaN in pandas

In questa conversione, una colonna Spark di interi che ammette null diventa una colonna pandas in virgola mobile con NaN al posto del valore mancante. La rappresentazione dei valori mancanti dipende dal tipo risultante in pandas.

menu_pd = pd.DataFrame({"drink_name": ["tea", "coffee"], "unit_price_text": ["2.50", "3.75"]})
menu = spark.createDataFrame(menu_pd, "drink_name STRING, unit_price_text STRING")
menu.show()

quantity_pd = quantities.toPandas()
print(quantity_pd)
assert quantity_pd["quantity"].isna().sum() == 1

19. Creare una tabella da un RDD

Un RDD è una collezione distribuita. In PySpark classico, spark.sparkContext dà accesso allo SparkContext. Qui parallelize() trasforma una piccola lista Python in un RDD e la stessa stringa DDL ne descrive le righe.

quantity_rdd = spark.sparkContext.parallelize(quantity_rows)
from_rdd = spark.createDataFrame(quantity_rdd, "order_id INT, quantity INT")
assert from_rdd.count() == 2

20. Possibilità di null in una colonna

StructField descrive un campo. nullable=True consente valori null. Qui order_id è obbligatorio e quantity può essere sconosciuta.

21. nullable=False non corregge i dati

Con questa lista Python e la verifica dello schema attiva, PySpark rifiuta None in un campo che non ammette null. Se la regola richiede un ID d’ordine, gestisci il valore mancante prima di creare il DataFrame.

quantity_schema = StructType([
    StructField("order_id", IntegerType(), nullable=False),
    StructField("quantity", IntegerType(), nullable=True),
])
spark.createDataFrame(quantity_rows, quantity_schema).printSchema()

try:
    spark.createDataFrame([(None, 2)], quantity_schema)
except PySparkValueError as error:
    assert error.getCondition() == "FIELD_NOT_NULLABLE_WITH_NAME"
    print(error.getCondition())
else:
    raise AssertionError("A missing order ID should fail schema verification.")

22. Tipi di dato di Spark

Tra i tipi comuni ci sono STRING, BOOLEAN, INT, BIGINT, DOUBLE, DECIMAL, DATE e TIMESTAMP. Una data rappresenta un giorno di calendario. Un timestamp contiene data e ora. L’esempio con il timestamp usa un fuso orario di sessione fisso.

23. Schema esplicito

Fornisci a Spark i nomi e i tipi dei campi quando fanno parte del contratto dei dati.

24. Schema dedotto

Se passi soltanto i nomi delle colonne, Spark deduce i tipi. Lo stesso intero Python diventa in genere BIGINT con la deduzione e INT se lo dichiari come tale.

spark.conf.set("spark.sql.session.timeZone", "UTC")
sample = [(101, "tea", True, 2.5, date(2026, 11, 4), datetime(2026, 11, 4, 9))]
names = ["order_id", "drink_name", "is_hot", "rating", "order_date", "ordered_at"]
typed = spark.createDataFrame(
    sample,
    "order_id INT, drink_name STRING, is_hot BOOLEAN, rating DOUBLE, order_date DATE, ordered_at TIMESTAMP_NTZ",
)
inferred = spark.createDataFrame(sample, names)
typed.printSchema()
inferred.printSchema()
assert typed.schema["order_id"].dataType.simpleString() == "int"
assert inferred.schema["order_id"].dataType.simpleString() == "bigint"

25. Convertire il tipo di una colonna

Usa cast() per convertire una colonna esistente. Il risultato qui sotto è di tipo intero, anche se i valori originali sono stringhe.

26. Modalità ANSI

L’impostazione di sessione spark.sql.ansi.enabled controlla come Spark gestisce alcune operazioni non valide. Impostala esplicitamente per gli esempi che seguono.

raw_quantities = spark.createDataFrame([("2",), ("oops",), (None,)], "quantity_text STRING")
valid_quantity = spark.createDataFrame([("2",)], "quantity_text STRING")
valid_quantity.select(col("quantity_text").cast("int").alias("quantity")).show()
spark.conf.set("spark.sql.ansi.enabled", "true")

27. Decimal per il denaro

Per gli importi monetari usa tipi decimali a virgola fissa. DECIMAL(8,2) consente otto cifre totali, di cui due dopo la virgola. Crea il Decimal Python da una stringa per preservare il valore desiderato.

money_schema = StructType([StructField("unit_price", DecimalType(8, 2), nullable=False)])
spark.createDataFrame([(Decimal("2.50"),)], money_schema).show()
menu.select("drink_name", col("unit_price_text").cast("decimal(8,2)").alias("unit_price")).show()

28. Conversione non valida in modalità ANSI

Il testo oops non si può convertire in un intero. L’errore emerge quando show() valuta la conversione. L’esempio intercetta l’errore previsto e permette a Run All di proseguire.

try:
    raw_quantities.select(col("quantity_text").cast("int").alias("quantity")).show()
except PySparkException as error:
    assert error.getCondition() == "CAST_INVALID_INPUT"
    print(error.getCondition())
else:
    raise AssertionError("The malformed quantity should fail in ANSI mode.")

29. Conversione sicura

In questa conversione supportata, try_cast() restituisce null per il testo malformato. Conserva il valore originale per distinguere il testo malformato da un input mancante. Qui isNull() verifica se il risultato manca e isNotNull() verifica se l’input è presente. lit() aggiunge un’etichetta costante per la provenienza.

parsed = raw_quantities.select(
    "quantity_text", col("quantity_text").try_cast("int").alias("quantity")
)
checked = parsed.select(
    "quantity_text",
    "quantity",
    (col("quantity_text").isNotNull() & col("quantity").isNull()).alias("invalid_quantity"),
    lit("cafe_demo").alias("source"),
)
checked.show()
assert {row.quantity_text: (row.quantity, row.invalid_quantity) for row in checked.collect()} == {
    "2": (2, False),
    "oops": (None, True),
    None: (None, False),
}

Verifica la comprensione

1. Quale schema assegna il tipo INT a order_id?
2. Che cosa conviene conservare quando una conversione sicura restituisce null per un input malformato?
3. Perché toPandas() è adatto all’esempio di due righe?

Risposte

1. Lo schema esplicito che dichiara order_id INT o usa IntegerType().
2. Conserva il testo originale o un indicatore della conversione fallita.
3. Entrambe le righe entrano facilmente nella memoria del driver.

Riferimenti

Documentazione API di Apache Spark per createDataFrame e Column.try_cast e guida alla conformità ANSI di Spark SQL.