{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 25 · Usare le API di basso livello\n\nEsplora 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.",
      "id": "cell-01"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Apache Arrow\n\nArrow è 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.",
      "id": "cell-02"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from operator import add\n\nfrom pyspark.sql import SparkSession\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"25-low-level-apis\").getOrCreate()\nsc = spark.sparkContext\nsales = spark.createDataFrame(\n    [(101, \"tea\", 2), (102, \"cake\", 1), (103, \"tea\", 3)],\n    \"order_id INT, item STRING, units INT\",\n)\narrow_table = sales.orderBy(\"order_id\").toArrow()\nassert arrow_table.column(\"units\").to_pylist() == [2, 1, 3]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-03"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 2. RDD\n\nUn RDD è una collezione partizionata e resiliente che Spark può calcolare tramite più task. Può contenere oggetti Python e non ha uno schema SQL.\n\n## 3. Dataset\n\nScala 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.\n\n## 4. Variabile broadcast\n\n`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.\n\n## 5. Accumulatore\n\nI 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.",
      "id": "cell-04"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "records = sc.parallelize([(\"tea\", 2), (\"cake\", 1), (\"tea\", 3)], 2)\nlabels = sc.broadcast({\"tea\": \"drink\", \"cake\": \"food\"})\nobserved = sc.accumulator(0)\nassert records.getNumPartitions() == 2\nassert observed.value == 0",
      "execution_count": null,
      "outputs": [],
      "id": "cell-05"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Trasformazioni su RDD\n\n`map()` e `filter()` descrivono nuovi RDD. Un’azione come `collect()` esegue le trasformazioni necessarie.\n\n## 7. Funzioni Python negli RDD\n\nSpark 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.\n\n## 8. Da DataFrame a RDD\n\nPySpark 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.\n\n## 9. Aggregare per chiave\n\n`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.\n\n## 10. Raggruppare i valori per chiave\n\n`groupByKey()` restituisce ogni chiave e l’iterabile di tutti i suoi valori.\n\n## 11. groupByKey su una chiave con molti valori\n\n`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.",
      "id": "cell-06"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def as_pair(row):\n    return row.item, row.units\n\npairs = sales.rdd.map(as_pair).filter(lambda pair: pair[1] > 0)\ntotals = pairs.reduceByKey(add, numPartitions=2)\nassert dict(totals.collect()) == {\"tea\": 5, \"cake\": 1}\ngroups = dict(pairs.groupByKey().mapValues(lambda values: sorted(values)).collect())\nassert groups == {\"tea\": [2, 3], \"cake\": [1]}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-07"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 12. Aggiornare un accumulatore in una trasformazione\n\nUna 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.\n\n## 13. Eseguire una funzione su ogni riga\n\n`foreach()` esegue una funzione su ciascun elemento nei task. Il valore restituito viene scartato. Gli effetti esterni devono poter tollerare nuovi tentativi.\n\n## 14. Accumulatore e join con trasmissione\n\nOgni 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.",
      "id": "cell-08"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def enrich(pair):\n    item, units = pair\n    return item, labels.value.get(item), units\n\ndef note_missing(row):\n    if row[1] is None:\n        observed.add(1)\n\ndef instrumented(pair):\n    mapped_in_transform.add(1)\n    return pair\n\ntry:\n    enriched = sc.parallelize([(\"tea\", 2), (\"water\", 1)], 2).map(enrich)\n    enriched.foreach(note_missing)\n    assert observed.value == 1\n\n    mapped_in_transform = sc.accumulator(0)\n    mapped = records.map(instrumented)\n    mapped.count()\n    print(\"After first action:\", mapped_in_transform.value)\n    mapped.count()\n    print(\"After second action:\", mapped_in_transform.value)\nfinally:\n    labels.destroy()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-09"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Py4J\n\nPySpark 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.",
      "id": "cell-10"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Spark version from the classic session:\", spark.version)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-11"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché usare reduceByKey per sommare quantità?\n2. Perché evitare accumulatori nelle trasformazioni per totali esatti?\n3. Dove conserva il risultato toArrow?\n\n### Risposte\n\n1. Spark può combinare somme parziali prima dello shuffle.\n2. La trasformazione può essere eseguita più volte.\n3. Nella memoria del driver come tabella PyArrow locale.",
      "id": "cell-12"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}