{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 06 · Scegliere e ordinare le righe\n\nFiltra, ordina e campiona righe d’ordine. La lezione autonoma usa PySpark classico 4.x e Java 17 o successivo. Su Kaggle avvia una nuova sessione CPU. Se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"` in una cella di preparazione, poi usa Run All.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.errors import PySparkValueError\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"06-choose-and-sort-rows\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nlines = spark.createDataFrame(\n    [(1, \"tea\", 2), (2, \"coffee\", 5), (3, \"iced tea\", None),\n     (4, \"cake\", 3), (5, \"tea\", 5), (6, None, 1)],\n    \"line_id INT, item STRING, quantity INT\",\n)\nitem = F.col(\"item\")\nqty = F.col(\"quantity\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Ordinare le righe\n\n`orderBy()` ordina l’intero risultato. Aggiungi una chiave univoca di spareggio per rendere riproducibile l’ordine tra valori uguali.\n\n## 2. Condizione\n\nUna condizione è un’espressione di colonna che restituisce vero, falso o sconosciuto. Qui verifica se la quantità è almeno tre.\n\n## 3. Filtrare le righe\n\n`filter()` conserva solo le righe per cui la condizione è vera. La quantità sconosciuta viene quindi esclusa.\n\n## 4. Confrontare valori\n\nUsa espressioni come `qty >= 3`, `qty == 5` e `qty != 5`. L’uguaglianza usa due segni di uguale.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "lines.orderBy(\"line_id\").show()\nat_least_three = qty >= 3\nassert lines.filter(at_least_three).count() == 3\nlines.filter(qty == 5).orderBy(\"line_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 5. Appartenenza a una lista\n\n`isin()` verifica se il valore della colonna coincide con uno di quelli forniti.\n\n## 6. Condizioni sul testo\n\n`contains()` cerca una sottostringa letterale. `startswith()` e `endswith()` verificano l’inizio e la fine della stringa. Gli esempi usano i valori minuscoli dei dati di prova.\n\n## 7. between include gli estremi\n\n`between(2, 5)` include sia 2 e 5 sia i valori interni all’intervallo.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert lines.filter(item.isin(\"tea\", \"coffee\")).count() == 3\nassert lines.filter(item.contains(\"tea\")).count() == 3\nassert lines.filter(item.startswith(\"tea\")).count() == 2\nassert lines.filter(item.endswith(\"tea\")).count() == 3\nassert lines.filter(qty.between(2, 5)).count() == 4",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 8. Controllare se un valore è null\n\nUsa `isNull()` per i valori mancanti e `isNotNull()` per quelli presenti.\n\n## 9. Confrontare con null\n\nI confronti ordinari con null danno sconosciuto, anche tra due null.\n\n## 10. Confrontare trattando i null come uguali\n\n`eqNullSafe()` restituisce vero per due null, falso quando uno solo è null e l’uguaglianza ordinaria per i valori presenti di questo esempio.\n\n## 11. == None non trova righe\n\n`qty == None` crea un confronto ordinario con null. Ogni esito è sconosciuto e il filtro conserva zero righe.\n\n## 12. Il confronto di disuguaglianza esclude i null\n\n`qty != 5` mantiene i valori presenti diversi da cinque. Per includere le quantità sconosciute, aggiungi una condizione sui null.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert lines.filter(qty.isNull()).count() == 1\nassert lines.filter(qty.isNotNull()).count() == 5\nassert lines.filter(qty.eqNullSafe(None)).count() == 1\nassert lines.filter(qty == None).count() == 0\nassert lines.filter(qty != 5).count() == 3\nassert lines.filter((qty != 5) | qty.isNull()).count() == 4\nlines.select(\"line_id\", (qty == 5).alias(\"equals_five\")).orderBy(\"line_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Combinare condizioni\n\nUsa `&` per AND, `|` per OR e `~` per NOT sulle colonne Spark. Negare un esito sconosciuto dà ancora sconosciuto.\n\n## 14. and, or e not di Python sulle colonne\n\nGli operatori booleani Python cercano di valutare l’oggetto Column stesso. Spark richiede invece un’espressione da valutare per ogni riga. L’esempio intercetta l’errore e prosegue.\n\n## 15. Precedenza di & senza parentesi\n\nPython interpreta `qty >= 2 & qty <= 5` come un confronto concatenato che coinvolge `2 & qty`. Racchiudi ogni confronto tra parentesi.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "condition = (qty >= 2) & (qty <= 5)\nassert lines.filter(condition).count() == 4\nassert lines.filter(~(qty == 5)).count() == 3\ntry:\n    (qty >= 2) and (qty <= 5)\nexcept PySparkValueError as error:\n    assert error.getCondition() == \"CANNOT_CONVERT_COLUMN_INTO_BOOL\"\n    print(error.getCondition())\nelse:\n    raise AssertionError(\"Expected Python boolean conversion to fail.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 16. Campione casuale\n\n`sample()` usa una frazione. La dimensione effettiva può differire dal prodotto tra frazione e numero di righe. Usa un seme per ripetere il campionamento a parità di input e disposizione delle partizioni.\n\n## 17. Campione stratificato\n\n`sampleBy()` accetta una frazione per strato. Qui gli ID pari e dispari formano due strati. Gli strati omessi hanno frazione zero. Le frazioni fissano i tassi di campionamento, mentre i conteggi effettivi possono variare.\n\n## 18. Dividere le righe in parti casuali\n\n`randomSplit()` normalizza i pesi relativi forniti. I pesi indicano le proporzioni attese. Controlla i conteggi effettivi.\n\n## 19. randomSplit senza cache\n\nAzioni separate possono rivalutare l’input comune a monte. Con input e partizioni stabili, un seme fisso è utile. Se il lavoro a monte può cambiare, dividi a partire da uno snapshot stabile. Materializzare la cache può evitare il ricalcolo finché i blocchi rimangono disponibili. Usa uno snapshot persistente per fissare l’input tra tentativi o sessioni.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "population = spark.range(200, numPartitions=2).withColumn(\"stratum\", F.col(\"id\") % 2)\nprint(\"Sample size:\", population.sample(withReplacement=False, fraction=0.2, seed=17).count())\nstratified = population.sampleBy(\"stratum\", fractions={0: 0.10, 1: 0.40}, seed=17)\nstratified.groupBy(\"stratum\").count().orderBy(\"stratum\").show()\n\ntrain, test = population.randomSplit([0.8, 0.2], seed=17)\ntrain_ids = {row.id for row in train.select(\"id\").collect()}\ntest_ids = {row.id for row in test.select(\"id\").collect()}\nassert train_ids.isdisjoint(test_ids)\nassert train_ids | test_ids == set(range(200))\nprint(\"Split sizes:\", len(train_ids), len(test_ids))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Ordinare in modo decrescente\n\n`desc()` mette prima i valori presenti più grandi. Usa `line_id` per risolvere i pari merito sulla quantità.\n\n## 21. Posizione dei null\n\nPer impostazione predefinita, i null vengono prima nell’ordine crescente e dopo in quello decrescente. Puoi impostare la posizione con `asc_nulls_last()` o `desc_nulls_first()`.\n\n## 22. Ordinare dentro le partizioni\n\n`sortWithinPartitions()` ordina le righe dentro ogni partizione. I confini tra partizioni separano ancora le sequenze ordinate.\n\n## 23. Ordine delle righe non garantito\n\nUn’anteprima può sembrare stabile tra esecuzioni. Se la logica a valle dipende dall’ordine, applica un ordinamento finale esplicito con chiavi di spareggio.\n\n## 24. Ordinamento globale con rimescolamento\n\nUn ordinamento globale su più partizioni può redistribuire le righe in intervalli ordinati. Nel piano dell’esempio cerca uno scambio con partizionamento per intervalli e un ordinamento.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "descending = lines.orderBy(qty.desc(), \"line_id\")\nassert [row.line_id for row in descending.collect()] == [2, 5, 4, 1, 6, 3]\nnull_first = lines.orderBy(qty.desc_nulls_first(), \"line_id\")\nassert null_first.first().line_id == 3\nlines.orderBy(qty.asc_nulls_last(), \"line_id\").show()\n\nlocal_sort = population.repartition(2, \"stratum\").sortWithinPartitions(F.col(\"id\").desc())\nlocal_sort.explain()\npopulation.orderBy(F.col(\"stratum\").desc(), F.col(\"id\").desc()).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 25. Campionare esattamente n righe\n\nSu una tabella piccola, assegna a ogni riga una chiave pseudocasuale con `rand(seed)`, ordina per chiave e prendi le prime n righe. Ottieni n righe se sono disponibili. Un ID univoco risolve i pari merito. I numeri casuali tornano nel Notebook 07. Il metodo valuta una chiave su tutto l’input e può essere costoso su larga scala.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "exact_sample = population.orderBy(F.rand(17), \"id\").limit(5)\nassert exact_sample.count() == 5\nexact_sample.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale espressione conserva le quantità sconosciute e quelle diverse da cinque?\n2. Perché un campione al 20% può avere un numero di righe diverso dal 20% dell’input?\n3. Quale ordinamento serve per ottenere un ordine unico tra tutte le partizioni?\n\n### Risposte\n\n1. `(qty != 5) | qty.isNull()`.\n2. La frazione determina un tasso di campionamento. La dimensione effettiva varia.\n3. Usa `orderBy()` e aggiungi chiavi di spareggio quando conta l’ordine tra pari merito.",
      "id": "cell-17"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}