{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 18 · Usare SQL\n\nInterroga le stesse vendite del bar con SQL e con l’API DataFrame. Usa una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa con `%pip install \"pyspark>=4.0,<4.1\"`, poi usa Run All. Tutti i dati sono nel notebook.",
      "id": "cell-01"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Spark SQL\n\nSpark SQL interroga dati strutturati. Qui ogni riga rappresenta una vendita e i prezzi sono espressi in centesimi interi.\n\n## 2. Registrare una vista temporanea\n\n`createOrReplaceTempView()` assegna a un DataFrame un nome SQL nella sessione corrente.\n\n## 3. Eseguire una query SQL\n\n`spark.sql()` restituisce un DataFrame per una SELECT. Usa `show()` per mostrare il risultato e `ORDER BY` quando conta l’ordine.",
      "id": "cell-02"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from uuid import uuid4\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"18-use-sql\").getOrCreate()\nsales = spark.createDataFrame(\n    [(101, \"tea\", 2, 250), (102, \"coffee\", 1, 375), (103, \"tea\", 4, 250), (104, None, 3, 100)],\n    \"sale_id INT, item STRING, units INT, price_cents INT\",\n)\nview = \"nb18_sales_\" + uuid4().hex[:8]\nsales.createOrReplaceTempView(view)\nspark.sql(f\"SELECT sale_id, item, units FROM {view} ORDER BY sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-03"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 4. Un solo motore per SQL e DataFrame\n\nEntrambe le interfacce usano il motore di query Spark. Usa `explain()` per esaminare i piani. Query equivalenti possono mostrare dettagli diversi, come gli ID delle espressioni.\n\n## 5. Selezionare con un’espressione SQL\n\n`selectExpr()` accetta un’espressione SQL per colonna di output. Usa `AS` per nominare i campi calcolati.\n\n## 6. Scrivere una condizione come testo SQL\n\n`filter()` accetta un predicato SQL come stringa o un’espressione Column. Gli esempi usano predicati fissi. Per valori esterni nelle query usa parametri.\n\n## 7. Condizione in SQL\n\nUsa `WHERE` per conservare le righe per cui il predicato è vero. Un confronto con un prodotto null dà sconosciuto, quindi la riga non passa.\n\n## 8. Espressione SQL in una colonna\n\n`expr()` interpreta un’espressione SQL come Column. Usala dentro metodi come `withColumn()`.",
      "id": "cell-04"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "query = spark.sql(f'''\n    SELECT sale_id, units * price_cents AS revenue_cents\n    FROM {view}\n    WHERE item = 'tea' AND units >= 2\n''')\nselected = sales.filter(\"item = 'tea' AND units >= 2\").selectExpr(\n    \"sale_id\", \"units * price_cents AS revenue_cents\"\n)\ncomputed = (\n    sales.filter((F.col(\"item\") == \"tea\") & (F.col(\"units\") >= 2))\n    .withColumn(\"revenue_cents\", F.expr(\"units * price_cents\"))\n    .select(\"sale_id\", \"revenue_cents\")\n)\nquery.explain()\nselected.explain()\nquery.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-05"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. Vista temporanea legata alla sessione\n\nUna nuova sessione Spark può condividere lo SparkContext e avere viste temporanee locali separate. Verifica la visibilità in entrambe le sessioni.\n\n## 10. Controllare se due query sono equivalenti\n\nConfronta gli schemi, allinea le colonne e usa `exceptAll()` in entrambe le direzioni per confrontare le righe, compresi i conteggi dei duplicati. Il controllo vale per questi dati di prova e non dimostra l’equivalenza su tutti gli input.",
      "id": "cell-06"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert spark.catalog.tableExists(view)\nassert not spark.newSession().catalog.tableExists(view)\n\ndef check_same_rows(left, right):\n    assert left.schema == right.schema\n    assert left.exceptAll(right).isEmpty()\n    assert right.exceptAll(left).isEmpty()\n\ncheck_same_rows(query, selected)\ncheck_same_rows(query, computed)\nassert {r.sale_id: r.revenue_cents for r in query.collect()} == {101: 500, 103: 1000}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-07"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. Parametri con nome in una query\n\nScrivi un segnaposto con nome come `:item_name` e passa il valore nel dizionario `args`. Il valore viene associato senza essere inserito nel testo della query.\n\n## 12. DataFrame come argomento di una query\n\nL’argomento con nome in `spark.sql(..., source=sales)` permette a Spark di inserire un riferimento al DataFrame. Usa un segnaposto come `{source}` dove serve una relazione. I segnaposto con nome continuano ad associare i valori separatamente.\n\n## 13. Iniezione SQL\n\nInserire valori grezzi dell’utente nel testo SQL permette loro di alterare la struttura della query. Associa i valori con segnaposto. Per identificativi selezionabili dall’utente, risolvili usando un elenco di identificativi ammessi. Qui si associa testo con apici e caratteri simili a SQL e si verifica che non corrisponda ad alcun prodotto.",
      "id": "cell-08"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "params = {\"item_name\": \"tea\", \"minimum_units\": 2}\nsql_template = '''\n    SELECT sale_id, units * price_cents AS revenue_cents\n    FROM {source}\n    WHERE item = :item_name AND units >= :minimum_units\n'''\nparameterized = spark.sql(sql_template, args=params, source=sales)\ncheck_same_rows(query, parameterized)\nliteral_only = spark.sql(\n    sql_template,\n    args={\"item_name\": \"tea' OR 1=1 --\", \"minimum_units\": 2},\n    source=sales,\n)\nassert literal_only.isEmpty()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-09"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Sintassi SQL a pipe\n\nSpark 4 supporta gli operatori SQL a pipe. Parti da una relazione e applica le fasi nell’ordine scritto con `|>`. La query filtra le vendite, seleziona campi calcolati e ordina l’output.",
      "id": "cell-10"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "piped = spark.sql(f'''\n    FROM {view}\n    |> WHERE item = 'tea' AND units >= 2\n    |> SELECT sale_id, units * price_cents AS revenue_cents\n    |> ORDER BY sale_id\n''')\ncheck_same_rows(query, piped)\nassert [r.sale_id for r in piped.collect()] == [101, 103]\npiped.show()\nspark.catalog.dropTempView(view)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-11"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale API trasforma un’espressione SQL testuale in una Column?\n2. Perché confrontare i risultati in entrambe le direzioni con exceptAll?\n3. Come dovresti fornire un nome di prodotto proveniente dall’esterno?\n\n### Risposte\n\n1. Usa `F.expr()`.\n2. Ogni direzione rileva righe o occorrenze duplicate presenti da un solo lato.\n3. Associalo come valore con un parametro con nome.",
      "id": "cell-12"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}