25 · Usare le API di basso livello

Esplora Arrow, RDD, variabili broadcast e accumulatori sotto l’API DataFrame. Usa PySpark classico 4.0.x e Java 17+. In una nuova sessione CPU su Kaggle, installa le dipendenze mancanti con %pip install "pyspark[sql]>=4.0,<4.1", poi usa Run All.

1. Apache Arrow

Arrow è un formato di dati colonnare. Spark può usarlo per trasferimenti tra JVM e Python. Spark 4 aggiunge toArrow(), che raccoglie un DataFrame in una tabella PyArrow locale. Usalo solo se il risultato entra nella memoria del driver.

from operator import add

from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[2]").appName("25-low-level-apis").getOrCreate()
sc = spark.sparkContext
sales = spark.createDataFrame(
    [(101, "tea", 2), (102, "cake", 1), (103, "tea", 3)],
    "order_id INT, item STRING, units INT",
)
arrow_table = sales.orderBy("order_id").toArrow()
assert arrow_table.column("units").to_pylist() == [2, 1, 3]

2. RDD

Un RDD è una collezione partizionata e resiliente che Spark può calcolare tramite più task. Può contenere oggetti Python e non ha uno schema SQL.

3. Dataset

Scala e Java espongono un’API Dataset tipizzata con encoder. PySpark espone DataFrame con uno schema disponibile durante l’esecuzione e RDD per operazioni Python di basso livello.

4. Variabile broadcast

sc.broadcast() distribuisce un valore di sola lettura che i task possono riutilizzare. I task vi accedono tramite .value. È una variabile condivisa, distinta da un broadcast join SQL.

5. Accumulatore

I task possono aggiungere valori a un accumulatore, mentre il driver ne legge il valore. Usalo per la diagnostica. La lezione mostra gli aggiornamenti in un’azione successiva.

records = sc.parallelize([("tea", 2), ("cake", 1), ("tea", 3)], 2)
labels = sc.broadcast({"tea": "drink", "cake": "food"})
observed = sc.accumulator(0)
assert records.getNumPartitions() == 2
assert observed.value == 0

6. Trasformazioni su RDD

map() e filter() descrivono nuovi RDD. Un’azione come collect() esegue le trasformazioni necessarie.

7. Funzioni Python negli RDD

Spark serializza la funzione Python e lo stato che le serve. I task la eseguono nei worker Python. Evita di catturare una SparkSession o un oggetto grande del driver.

8. Da DataFrame a RDD

PySpark classico espone df.rdd come RDD di oggetti Row. Spostare il calcolo successivo in Python ti dà controllo sui record, ma fornisce a Catalyst meno informazioni sulla logica.

9. Aggregare per chiave

reduceByKey() combina i valori con una funzione associativa e commutativa. Spark può combinare i valori locali prima dello shuffle. L’addizione è adatta a questi totali interi.

10. Raggruppare i valori per chiave

groupByKey() restituisce ogni chiave e l’iterabile di tutti i suoi valori.

11. groupByKey su una chiave con molti valori

groupByKey() raggruppa i singoli valori. Una chiave molto frequente può produrre un iterabile grande e più traffico di shuffle. Usa reduceByKey() quando serve solo un aggregato che può essere combinato.

def as_pair(row):
    return row.item, row.units

pairs = sales.rdd.map(as_pair).filter(lambda pair: pair[1] > 0)
totals = pairs.reduceByKey(add, numPartitions=2)
assert dict(totals.collect()) == {"tea": 5, "cake": 1}
groups = dict(pairs.groupByKey().mapValues(lambda values: sorted(values)).collect())
assert groups == {"tea": [2, 3], "cake": [1]}

12. Aggiornare un accumulatore in una trasformazione

Una trasformazione può essere eseguita di nuovo durante un’azione successiva o un ricalcolo. I suoi aggiornamenti all’accumulatore sono inadatti a totali esatti di business. Le due azioni sotto valutano intenzionalmente due volte la stessa trasformazione senza cache.

13. Eseguire una funzione su ogni riga

foreach() esegue una funzione su ciascun elemento nei task. Il valore restituito viene scartato. Gli effetti esterni devono poter tollerare nuovi tentativi.

14. Accumulatore e join con trasmissione

Ogni task accede al valore broadcast locale al worker per arricchire le righe, come in un piccolo join molti a uno. L’accumulatore conta le corrispondenze mancanti in un’azione separata. Con DataFrame, usa un broadcast join e calcola i conteggi dal suo risultato.

def enrich(pair):
    item, units = pair
    return item, labels.value.get(item), units

def note_missing(row):
    if row[1] is None:
        observed.add(1)

def instrumented(pair):
    mapped_in_transform.add(1)
    return pair

try:
    enriched = sc.parallelize([("tea", 2), ("water", 1)], 2).map(enrich)
    enriched.foreach(note_missing)
    assert observed.value == 1

    mapped_in_transform = sc.accumulator(0)
    mapped = records.map(instrumented)
    mapped.count()
    print("After first action:", mapped_in_transform.value)
    mapped.count()
    print("After second action:", mapped_in_transform.value)
finally:
    labels.destroy()

15. Py4J

PySpark classico usa Py4J per chiamare API JVM dal driver Python. L’esecuzione nei worker Python e i trasferimenti Arrow usano canali propri. Un errore lato Java può emergere tramite il ponte. Preferisci le API pubbliche PySpark ai riferimenti JVM privati.

print("Spark version from the classic session:", spark.version)

Verifica la comprensione

1. Perché usare reduceByKey per sommare quantità?
2. Perché evitare accumulatori nelle trasformazioni per totali esatti?
3. Dove conserva il risultato toArrow?

Risposte

1. Spark può combinare somme parziali prima dello shuffle.
2. La trasformazione può essere eseguita più volte.
3. Nella memoria del driver come tabella PyArrow locale.