{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 23 · Usare MLlib, Spark Connect e pandas API on Spark\n\nAddestra 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.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import os\n\nos.environ.setdefault(\"PYARROW_IGNORE_TIMEZONE\", \"1\")\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.ml import Pipeline\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator\nfrom pyspark.ml.feature import StandardScaler, VectorAssembler\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"23-spark-ecosystem\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\n\norders = spark.createDataFrame(\n    [(i, float(i % 12 + 1), float(i % 5 + 1), float(i % 12 >= 6)) for i in range(120)],\n    \"order_id INT, preparation_minutes DOUBLE, queue_length DOUBLE, late DOUBLE\",\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. MLlib\n\nMLlib è la libreria di apprendimento automatico di Spark. L’API DataFrame si aspetta colonne con nome per predittori ed etichetta.\n\n## 2. Trasformatore e stimatore\n\nUn trasformatore usa `transform()` per produrre un altro DataFrame. Uno stimatore usa `fit()` per apprendere un modello che può poi trasformare dati.\n\n## 3. Preparare le variabili del modello\n\n`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.\n\n## 4. Pipeline di apprendimento automatico\n\nUna pipeline concatena passaggi. Qui assembla i predittori, addestra un standardizzatore e addestra un modello di regressione logistica.\n\n## 5. Addestrare e valutare un classificatore\n\nDividi 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.\n\n## 6. Fuga di informazione\n\nAdatta 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.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "training = orders.filter(F.col(\"order_id\") < 96)\ntest = orders.filter(F.col(\"order_id\") >= 96)\nassembler = VectorAssembler(inputCols=[\"preparation_minutes\", \"queue_length\"], outputCol=\"raw_features\")\nscaler = StandardScaler(inputCol=\"raw_features\", outputCol=\"features\", withStd=True, withMean=False)\nclassifier = LogisticRegression(featuresCol=\"features\", labelCol=\"late\", maxIter=20, regParam=0.01)\nmodel = Pipeline(stages=[assembler, scaler, classifier]).fit(training)\npredictions = model.transform(test)\naccuracy = MulticlassClassificationEvaluator(\n    labelCol=\"late\", predictionCol=\"prediction\", metricName=\"accuracy\"\n).evaluate(predictions)\nassert predictions.count() == 24\nassert 0 <= accuracy <= 1\nprint(f\"Held-out accuracy: {accuracy:.3f}\")\npredictions.select(\"order_id\", \"late\", \"prediction\").orderBy(\"order_id\").show(5)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Spark Connect\n\nSpark 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.\n\n## 8. API non disponibili con Spark Connect\n\nIl 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.\n\n## 9. Integrazione con altri sistemi\n\nUn’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.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import subprocess\nimport sys\n\nendpoint = os.environ.get(\"NB23_SPARK_REMOTE\")\nif endpoint:\n    script = '''\nimport os\nfrom pyspark.sql import SparkSession\nremote = SparkSession.builder.remote(os.environ[\"NB23_SPARK_REMOTE\"]).getOrCreate()\ntry:\n    assert remote.range(3).count() == 3\n    print(\"Remote range count verified.\")\nfinally:\n    remote.stop()\n'''\n    environment = dict(os.environ)\n    for variable in (\"PYSPARK_GATEWAY_PORT\", \"PYSPARK_GATEWAY_SECRET\"):\n        environment.pop(variable, None)\n    subprocess.run([sys.executable, \"-c\", script], env=environment, check=True, timeout=60)\nelse:\n    print(\"Connect example skipped. Configure NB23_SPARK_REMOTE to run it.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. pandas API on Spark\n\nQuesta 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.\n\n## 11. Indice predefinito\n\nUn 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.\n\n## 12. Leggere un CSV con pandas-on-Spark\n\nUsa `ps.read_csv()` ed esamina le colonne risultanti. L’esempio scrive il proprio CSV.\n\n## 13. Convertire tra i tre tipi di DataFrame\n\nL’esempio usa `ps.from_pandas()`, `to_spark()`, `pandas_api()` e `to_pandas()`. La conversione in pandas porta i dati nella memoria locale.\n\n## 14. to_spark perde l’indice per impostazione predefinita\n\nUn 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`.\n\n## 15. PySpark e pandas a confronto\n\nI 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à.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nimport pandas as pd\nimport pyspark.pandas as ps\n\nansi = spark.conf.get(\"spark.sql.ansi.enabled\")\ntry:\n    spark.conf.set(\"spark.sql.ansi.enabled\", \"false\")\n    with ps.option_context(\"compute.default_index_type\", \"distributed\"):\n        with TemporaryDirectory(prefix=\"nb23-\") as directory:\n            csv = Path(directory) / \"orders.csv\"\n            csv.write_text(\"order_id,units\\n101,2\\n102,3\\n\", encoding=\"utf-8\")\n            ps_orders = ps.read_csv(str(csv))\n            assert set(ps_orders.columns) == {\"order_id\", \"units\"}\n            print(ps_orders.head(2))\n\n        local = pd.DataFrame({\"units\": [2, 3]}, index=pd.Index([101, 102], name=\"order_id\"))\n        distributed = ps.from_pandas(local)\n        assert distributed.to_spark().columns == [\"units\"]\n        explicit_index = distributed.to_spark(index_col=\"order_id\")\n        assert {r.order_id for r in explicit_index.collect()} == {101, 102}\n        round_trip = explicit_index.pandas_api(index_col=\"order_id\").to_pandas()\n        pd.testing.assert_frame_equal(round_trip.sort_index(), local.sort_index())\nfinally:\n    spark.conf.set(\"spark.sql.ansi.enabled\", ansi)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché adattare il standardizzatore dopo aver diviso i dati?\n2. Dove esegue la query Spark Connect?\n3. Come può sopravvivere un indice pandas alla conversione in un DataFrame Spark?\n\n### Risposte\n\n1. Usare solo i dati di addestramento evita che informazioni del test entrino nella preparazione.\n2. Il server Spark esegue il piano.\n3. Passa il nome della colonna indice a `to_spark(index_col=...)`.",
      "id": "cell-09"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}