23 · Usare MLlib, Spark Connect e pandas API on Spark

Addestra un classificatore e confronta tre interfacce DataFrame. Esegui in una nuova sessione CPU su Kaggle con Java 17+ e PySpark classico 4.0.x. Installa le dipendenze mancanti con %pip install "pyspark[pandas_on_spark,connect]>=4.0,<4.1" numpy. L’esempio Connect viene eseguito a parte solo se è configurato un endpoint.

import os

os.environ.setdefault("PYARROW_IGNORE_TIMEZONE", "1")

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import StandardScaler, VectorAssembler

spark = SparkSession.builder.master("local[2]").appName("23-spark-ecosystem").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")

orders = spark.createDataFrame(
    [(i, float(i % 12 + 1), float(i % 5 + 1), float(i % 12 >= 6)) for i in range(120)],
    "order_id INT, preparation_minutes DOUBLE, queue_length DOUBLE, late DOUBLE",
)

1. MLlib

MLlib è la libreria di apprendimento automatico di Spark. L’API DataFrame si aspetta colonne con nome per predittori ed etichetta.

2. Trasformatore e stimatore

Un trasformatore usa transform() per produrre un altro DataFrame. Uno stimatore usa fit() per apprendere un modello che può poi trasformare dati.

3. Preparare le variabili del modello

VectorAssembler combina predittori numerici in un vettore. Escludi etichetta e identificativo dell’ordine. Nell’esempio sintetico si assume che entrambi i predittori siano disponibili al momento della previsione.

4. Pipeline di apprendimento automatico

Una pipeline concatena passaggi. Qui assembla i predittori, addestra un standardizzatore e addestra un modello di regressione logistica.

5. Addestrare e valutare un classificatore

Dividi i dati prima dell’addestramento e valuta su un insieme tenuto da parte. Riporta dimensione del test e accuratezza. Le etichette sintetiche sono volutamente facili da separare, quindi il punteggio serve a verificare la pipeline.

6. Fuga di informazione

Adatta la preparazione soltanto alle righe di addestramento. Escludi campi derivati dall’esito o indisponibili al momento della previsione. Con dati reali usa una divisione temporale o per entità se le osservazioni dipendono tra loro.

training = orders.filter(F.col("order_id") < 96)
test = orders.filter(F.col("order_id") >= 96)
assembler = VectorAssembler(inputCols=["preparation_minutes", "queue_length"], outputCol="raw_features")
scaler = StandardScaler(inputCol="raw_features", outputCol="features", withStd=True, withMean=False)
classifier = LogisticRegression(featuresCol="features", labelCol="late", maxIter=20, regParam=0.01)
model = Pipeline(stages=[assembler, scaler, classifier]).fit(training)
predictions = model.transform(test)
accuracy = MulticlassClassificationEvaluator(
    labelCol="late", predictionCol="prediction", metricName="accuracy"
).evaluate(predictions)
assert predictions.count() == 24
assert 0 <= accuracy <= 1
print(f"Held-out accuracy: {accuracy:.3f}")
predictions.select("order_id", "late", "prediction").orderBy("order_id").show(5)

7. Spark Connect

Spark Connect separa client e server Spark. Il client invia piani non risolti tramite gRPC e riceve i risultati. In un processo Python separato, l’esempio opzionale si connette a un server già configurato. Usa l’endpoint fornito dall’ambiente.

8. API non disponibili con Spark Connect

Il client Connect non espone lo SparkContext classico né l’API RDD. Usa operazioni DataFrame e verifica il supporto delle singole API nelle versioni del client e del server.

9. Integrazione con altri sistemi

Un’applicazione remota può inviare query DataFrame con Connect. L’accesso a storage e database avviene dall’ambiente del server Spark, che deve disporre del connettore e degli accessi necessari. L’esempio legge solo una sequenza generata.

import subprocess
import sys

endpoint = os.environ.get("NB23_SPARK_REMOTE")
if endpoint:
    script = '''
import os
from pyspark.sql import SparkSession
remote = SparkSession.builder.remote(os.environ["NB23_SPARK_REMOTE"]).getOrCreate()
try:
    assert remote.range(3).count() == 3
    print("Remote range count verified.")
finally:
    remote.stop()
'''
    environment = dict(os.environ)
    for variable in ("PYSPARK_GATEWAY_PORT", "PYSPARK_GATEWAY_SECRET"):
        environment.pop(variable, None)
    subprocess.run([sys.executable, "-c", script], env=environment, check=True, timeout=60)
else:
    print("Connect example skipped. Configure NB23_SPARK_REMOTE to run it.")

10. pandas API on Spark

Questa API offre operazioni in stile pandas eseguite tramite Spark. In Spark 4.0 il controllo di compatibilità predefinito può rifiutare una sessione con ANSI abilitata. L’esempio disattiva temporaneamente ANSI nella sessione e poi ripristina l’impostazione.

11. Indice predefinito

Un DataFrame pandas-on-Spark ha un indice. Un indice generato automaticamente non stabilisce l’identità di business. Qui un indice distribuito evita di imporre una numerazione sequenziale delle righe.

12. Leggere un CSV con pandas-on-Spark

Usa ps.read_csv() ed esamina le colonne risultanti. L’esempio scrive il proprio CSV.

13. Convertire tra i tre tipi di DataFrame

L’esempio usa ps.from_pandas(), to_spark(), pandas_api() e to_pandas(). La conversione in pandas porta i dati nella memoria locale.

14. to_spark perde l’indice per impostazione predefinita

Un DataFrame Spark non ha un indice pandas. Passa index_col per conservare i valori dell’indice come colonna ordinaria. Qui l’indice di origine si chiama esplicitamente order_id.

15. PySpark e pandas a confronto

I DataFrame Spark sono distribuiti e non hanno un ordine implicito delle righe. I DataFrame pandas sono locali e usano le etichette dell’indice per l’allineamento. pandas API on Spark offre una sintassi familiare, ma può svolgere lavoro distribuito o presentare differenze di compatibilità.

from pathlib import Path
from tempfile import TemporaryDirectory

import pandas as pd
import pyspark.pandas as ps

ansi = spark.conf.get("spark.sql.ansi.enabled")
try:
    spark.conf.set("spark.sql.ansi.enabled", "false")
    with ps.option_context("compute.default_index_type", "distributed"):
        with TemporaryDirectory(prefix="nb23-") as directory:
            csv = Path(directory) / "orders.csv"
            csv.write_text("order_id,units\n101,2\n102,3\n", encoding="utf-8")
            ps_orders = ps.read_csv(str(csv))
            assert set(ps_orders.columns) == {"order_id", "units"}
            print(ps_orders.head(2))

        local = pd.DataFrame({"units": [2, 3]}, index=pd.Index([101, 102], name="order_id"))
        distributed = ps.from_pandas(local)
        assert distributed.to_spark().columns == ["units"]
        explicit_index = distributed.to_spark(index_col="order_id")
        assert {r.order_id for r in explicit_index.collect()} == {101, 102}
        round_trip = explicit_index.pandas_api(index_col="order_id").to_pandas()
        pd.testing.assert_frame_equal(round_trip.sort_index(), local.sort_index())
finally:
    spark.conf.set("spark.sql.ansi.enabled", ansi)

Verifica la comprensione

1. Perché adattare il standardizzatore dopo aver diviso i dati?
2. Dove esegue la query Spark Connect?
3. Come può sopravvivere un indice pandas alla conversione in un DataFrame Spark?

Risposte

1. Usare solo i dati di addestramento evita che informazioni del test entrino nella preparazione.
2. Il server Spark esegue il piano.
3. Passa il nome della colonna indice a to_spark(index_col=...).