05 · Scegliere le colonne

Seleziona, rinomina e calcola colonne mantenendo le righe di input. La lezione è autonoma e usa PySpark classico 4.x e Java 17 o successivo. In una nuova sessione CPU su Kaggle, se serve, installa PySpark in una cella di preparazione con %pip install "pyspark>=4.0,<4.1", poi usa Run All. I dati sono definiti qui sotto.

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

spark = SparkSession.builder.master("local[2]").appName("05-choose-columns").getOrCreate()
sales = spark.createDataFrame(
    [(101, " tea ", 2, 250, " counter "), (102, "cake", 1, 375, " takeaway ")],
    "order_id INT, item STRING, units INT, unit_price_cents INT, sales_channel STRING",
)
sales.show()

1. Selezionare colonne per nome

Passa i nomi a select() nell’ordine desiderato per l’output. L’anteprima conserva una riga per ogni riga d’ordine di partenza.

2. Rinominare nella selezione

Usa alias() su un’espressione di colonna per assegnarle il nome di output.

3. Indicare una colonna

F.col("units") crea un riferimento che Spark risolve rispetto all’input. sales["units"] fa riferimento a una colonna di questo DataFrame. Puoi passare una stringa a metodi che accettano nomi di colonne, come select().

sales.select("item", "units").show()
labels = sales.select(sales["item"].alias("product"), F.col("units").alias("quantity"))
assert labels.columns == ["product", "quantity"]
labels.show()

4. Rinominare tutte le colonne per posizione

toDF() assegna una lista completa di nomi alle posizioni delle colonne esistenti. Passa un nome per ogni colonna di input.

5. Eliminare colonne

drop() conserva le colonne rimanenti. Usalo quando vuoi eliminare alcuni campi noti.

6. Eliminare una colonna con un nome inesistente

Eliminare una colonna indicando un nome inesistente come stringa lascia lo schema com’era. Verifica che i campi di input attesi esistano se la loro assenza segnalerebbe un problema nei dati.

new_names = ["receipt_id", "product", "quantity", "price_cents", "channel"]
assert sales.toDF(*new_names).columns == new_names
assert sales.drop("sales_channel").columns == ["order_id", "item", "units", "unit_price_cents"]
assert sales.drop("missing_column").columns == sales.columns

7. Rinominare una colonna

withColumnRenamed() associa un nome esistente a un nome nuovo mantenendo le altre colonne.

8. Rinominare più colonne

withColumnsRenamed() accetta una corrispondenza tra nomi esistenti e nuovi nomi. Usa nomi di output distinti per evitare riferimenti ambigui.

9. Nome generato di una colonna calcolata

Spark può generare un nome di output da un’espressione. Assegna un alias all’espressione se altro codice dovrà riferirsi al risultato.

assert sales.withColumnRenamed("item", "product").columns[1] == "product"
renamed = sales.withColumnsRenamed({"item": "product", "units": "quantity"})
assert renamed.columns == ["order_id", "product", "quantity", "unit_price_cents", "sales_channel"]

line_total = F.col("units") * F.col("unit_price_cents")
print("Generated name:", sales.select(line_total).columns)
sales.select(line_total.alias("line_total_cents")).show()

10. toDF con colonne in ordine diverso

Quando cambia l’ordine di input, la stessa lista di nomi può etichettare i valori sbagliati. Qui entrambi i campi scambiati sono interi, quindi anche un controllo dei tipi mancherebbe l’errore. Rinomina per nome quando l’ordine di input può variare.

reordered = sales.select("units", "item", "order_id", "unit_price_cents", "sales_channel")
wrong = reordered.toDF(*new_names)
assert {row.receipt_id for row in wrong.select("receipt_id").collect()} == {1, 2}
fixed = reordered.withColumnsRenamed(dict(zip(sales.columns, new_names)))
assert {row.receipt_id for row in fixed.select("receipt_id").collect()} == {101, 102}

11. Selezionare colonne per tipo

Usa i campi dello schema per trovare le colonne con tipi numerici Spark. Tra queste ci sono anche gli identificativi numerici. Verifica il significato di ogni colonna prima di trattarla come misura.

12. Selezionare colonne con una regola sul nome

Filtra la lista dei nomi delle colonne con una regola Python sulle stringhe. Qui il suffisso _cents identifica le colonne espresse in centesimi.

13. Applicare la stessa operazione a più colonne

Costruisci un’espressione per ciascuna colonna da modificare. Qui trim() rimuove gli spazi ai bordi di due campi di testo. Il controllo con isin() verifica che le etichette dei canali ripulite siano nella lista dei valori ammessi. Le condizioni sul testo tornano nel Notebook 06.

numeric_names = [field.name for field in sales.schema if isinstance(field.dataType, NumericType)]
assert numeric_names == ["order_id", "units", "unit_price_cents"]
sales.select(*numeric_names).show()

cent_names = [name for name in sales.columns if name.endswith("_cents")]
assert cent_names == ["unit_price_cents"]

text_names = {"item", "sales_channel"}
cleaned = sales.select(*[
    F.trim(name).alias(name) if name in text_names else F.col(name)
    for name in sales.columns
])
cleaned.show()
valid_channel = F.col("sales_channel").isin("counter", "takeaway")
assert cleaned.filter(valid_channel).count() == 2

14. Nodo Project nel piano

Usa explain() per vedere come Spark rappresenta la selezione. Cerca un nodo Project che calcola le espressioni selezionate. Spark può semplificare le proiezioni, quindi il piano stampato dipende dalla query e dalla sua ottimizzazione. Il Notebook 20 tratta i piani in dettaglio.

cleaned.select("order_id", line_total.alias("line_total_cents")).explain()

15. Aggiungere più colonne insieme

Passa a withColumns() una corrispondenza tra nomi ed espressioni. Entrambe le espressioni qui sotto usano colonne già presenti nell’input. Se una seconda espressione richiede una colonna appena calcolata, rendi esplicita la dipendenza con un passaggio separato.

16. withColumn in un ciclo

Ogni chiamata a withColumn() aggiunge una proiezione al piano. Ripeterla molte volte può ingrandire il piano logico, anche se Spark in seguito combina le proiezioni. Per espressioni indipendenti, usa una sola chiamata a withColumns() o select(). Il confronto verifica che i due approcci diano gli stessi valori.

additions = {
    "line_total_cents": line_total,
    "large_line": F.col("units") >= 2,
}
one_step = cleaned.withColumns(additions)
loop = cleaned
for name, expression in additions.items():
    loop = loop.withColumn(name, expression)

comparison = ["order_id", "line_total_cents", "large_line"]
assert one_step.select(*comparison).orderBy("order_id").collect() == (
    loop.select(*comparison).orderBy("order_id").collect()
)
one_step.select(*comparison).orderBy("order_id").show()

Verifica la comprensione

1. Come puoi tenere solo item e rinominarla in product?
2. Perché riutilizzare una lista di nomi per posizione può etichettare male dati riordinati senza produrre un errore?
3. Come puoi aggiungere due colonne derivate indipendenti in una sola operazione?

Risposte

1. Usa sales.select(F.col("item").alias("product")).
2. toDF() assegna i nomi per posizione. I campi scambiati possono avere lo stesso tipo.
3. Passa a withColumns() una corrispondenza tra nomi ed espressioni.