{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 08 · Lavorare con date e ore\n\nInterpreta date, misura intervalli e raggruppa eventi in finestre temporali. La lezione usa PySpark classico 4.0.x e Java 17+. In una nuova sessione CPU su Kaggle, se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"`, poi usa Run All. Tutti i dati sono definiti nel notebook.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"08-dates-and-times\").getOrCreate()\nspark.conf.set(\"spark.sql.session.timeZone\", \"UTC\")\nspark.conf.set(\"spark.sql.timestampType\", \"TIMESTAMP_LTZ\")\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")\nspark.conf.set(\"spark.sql.legacy.timeParserPolicy\", \"CORRECTED\")\n\nraw = spark.createDataFrame(\n    [(101, \"2026-11-01\", \"2026-11-01 09:05\"),\n     (102, \"2026-11-02\", \"2026-11-02 12:30\"),\n     (103, None, \"bad input\")],\n    \"order_id INT, day_text STRING, time_text STRING\",\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Tipi date e timestamp\n\nDATE rappresenta un giorno di calendario. Un timestamp può includere ore, minuti, secondi e frazioni. Il notebook usa per impostazione predefinita timestamp che rappresentano istanti e parte con UTC come fuso di sessione.\n\n## 2. Convertire testo in timestamp\n\nUsa `to_timestamp()` con un formato di input quando sai che il testo è valido.\n\n## 3. Convertire testo in data\n\nUsa `to_date()` quando serve solo il giorno di calendario. Un input mancante resta null.\n\n## 4. Formato di una data\n\nIn `yyyy-MM-dd` i campi sono anno, mese e giorno. Il formato deve descrivere il testo di input.\n\n## 5. MM e mm\n\nI formati distinguono maiuscole e minuscole. `MM` indica il mese, `mm` i minuti e `HH` l’ora su 24 ore.\n\n## 6. Conversione sicura in timestamp\n\n`try_to_timestamp()` restituisce null per input malformati anche in modalità ANSI. Fornisci un formato valido come colonna letterale. Conserva il testo originale per distinguere valori malformati e mancanti.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "day_pattern = \"yyyy-MM-dd\"\ntime_pattern = \"yyyy-MM-dd HH:mm\"\nraw.filter(\"order_id < 103\").select(F.to_timestamp(\"time_text\", time_pattern)).show()\nparsed = raw.select(\n    \"*\",\n    F.to_date(\"day_text\", day_pattern).alias(\"order_day\"),\n    F.try_to_timestamp(\"time_text\", F.lit(time_pattern)).alias(\"order_time\"),\n)\nassert parsed.filter(F.col(\"order_time\").isNull()).count() == 1\nparsed.printSchema()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Estrarre parti di una data\n\nUsa funzioni come `year()`, `month()` e `dayofmonth()` per ricavare campi numerici.\n\n## 8. Formattare una data come testo\n\n`date_format()` produce una stringa da visualizzare o esportare. Conserva la colonna tipizzata per i calcoli.\n\n## 9. YYYY e yyyy\n\nUsa `yyyy` minuscolo per l’anno di calendario. In alcuni sistemi `Y` indica l’anno basato sulle settimane. I formati corretti di Spark rifiutano le lettere basate sulle settimane. La distinzione conta a cavallo di Capodanno.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "parsed.select(\n    \"order_day\", F.year(\"order_day\").alias(\"year\"), F.month(\"order_day\").alias(\"month\"),\n    F.dayofmonth(\"order_day\").alias(\"day\"), F.date_format(\"order_day\", \"dd/MM/yyyy\").alias(\"label\"),\n).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Troncare una data\n\nIl troncamento porta un valore all’inizio di un’unità. Troncare al mese dà il primo giorno del mese.\n\n## 11. dayofweek parte dalla domenica\n\n`dayofweek()` in Spark restituisce 1 per domenica e 7 per sabato.\n\n## 12. date_trunc dà un timestamp, trunc una data\n\nUsa `trunc(date, unit)` per un DATE e `date_trunc(unit, timestamp)` per un timestamp. L’ordine degli argomenti cambia.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "buckets = parsed.select(\n    \"order_id\", F.trunc(\"order_day\", \"month\").alias(\"month_date\"),\n    F.dayofweek(\"order_day\").alias(\"weekday\"),\n    F.date_trunc(\"month\", \"order_time\").alias(\"month_timestamp\"),\n)\nbuckets.printSchema()\nassert buckets.filter(\"order_id = 101\").first().weekday == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Differenze tra date\n\n`datediff(end, start)` restituisce giorni di calendario con segno. Opera sulle date e non misura le ore trascorse.\n\n## 14. Spostare una data\n\nUsa `date_add()` e `date_sub()` per spostarti di giorni di calendario.\n\n## 15. Creare un calendario\n\n`sequence()` restituisce un array, una raccolta memorizzata in una cella. Qui genera date con passo di un giorno ed estremi inclusi. `expr()` incorpora l’espressione SQL dell’intervallo e `explode()` emette una riga per elemento.\n\n## 16. add_months a fine mese\n\nQuando il mese di destinazione non contiene il giorno di partenza, `add_months()` usa il suo ultimo giorno valido.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "parsed.select(\n    F.datediff(\"order_day\", F.lit(\"2026-11-01\")).alias(\"days_since_start\"),\n    F.date_add(\"order_day\", 2).alias(\"follow_up\"),\n    F.date_sub(\"order_day\", 2).alias(\"two_days_before\"),\n).show()\ncalendar = spark.range(1).select(\n    F.explode(F.expr(\"sequence(DATE '2026-11-01', DATE '2026-11-07', INTERVAL 1 DAY)\")).alias(\"day\")\n)\nassert calendar.count() == 7\ncalendar.orderBy(\"day\").show()\nassert spark.sql(\"SELECT CAST(add_months(DATE '2024-01-31', 1) AS STRING) AS day\").first().day == \"2024-02-29\"",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 17. Finestra temporale fissa\n\nUna finestra fissa raggruppa gli eventi in intervalli senza sovrapposizioni. Spark include l’inizio ed esclude la fine. Un evento alle 09:10 appartiene all’intervallo che inizia alle 09:10.\n\n## 18. Finestra temporale scorrevole\n\nUna finestra di dieci minuti con uno scorrimento di cinque minuti si sovrappone alle finestre vicine. Uno stesso evento può comparire in più finestre.\n\n## 19. Sessione di eventi\n\nUna sessione raggruppa eventi vicini per la stessa chiave. Ogni evento estende la sessione candidata fino al proprio orario più la durata della pausa ammessa. Le candidate vicine si uniscono. Gli eventi dell’esempio evitano il limite esatto della pausa.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "events = spark.createDataFrame(\n    [(\"A\", \"2026-11-01 09:02\"), (\"A\", \"2026-11-01 09:06\"),\n     (\"A\", \"2026-11-01 09:10\"), (\"A\", \"2026-11-01 09:25\"),\n     (\"B\", \"2026-11-01 09:06\")],\n    \"customer STRING, event_text STRING\",\n).withColumn(\"event_time\", F.to_timestamp(\"event_text\", time_pattern))\n\nfixed = events.groupBy(F.window(\"event_time\", \"10 minutes\").alias(\"bucket\")).count()\nfixed.orderBy(\"bucket.start\").show(truncate=False)\nassert sorted(row[\"count\"] for row in fixed.collect()) == [1, 1, 3]\n\nsliding = events.groupBy(F.window(\"event_time\", \"10 minutes\", \"5 minutes\").alias(\"bucket\")).count()\nassert sliding.agg(F.sum(\"count\")).first()[0] == 10\nsliding.orderBy(\"bucket.start\").show(truncate=False)\n\nsessions = events.groupBy(\"customer\", F.session_window(\"event_time\", \"5 minutes\").alias(\"session\")).count()\nassert sorted((r.customer, r[\"count\"]) for r in sessions.collect()) == [(\"A\", 1), (\"A\", 3), (\"B\", 1)]\nsessions.orderBy(\"customer\", \"session.start\").show(truncate=False)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Fuso orario della sessione\n\nIl fuso della sessione influisce sull’interpretazione del testo senza fuso quando lo converti in un istante e sulla visualizzazione degli istanti. Per la riproducibilità, imposta un nome di regione IANA esplicito o UTC.\n\n## 21. Finestre allineate a UTC\n\nLe finestre fisse di Spark si allineano per impostazione predefinita all’epoca UTC. Un giorno in questa finestra equivale a 24 ore fisse. L’inizio può apparire a un’ora locale diversa da mezzanotte e lo scarto può cambiare con l’ora legale. Per raggruppare per giorno operativo locale, ricava una data di calendario locale.\n\n## 22. TIMESTAMP e TIMESTAMP_NTZ\n\nCon l’impostazione della lezione, TIMESTAMP equivale a TIMESTAMP_LTZ e rappresenta un istante visualizzato nel fuso della sessione. TIMESTAMP_NTZ conserva data e ora locali senza fuso. Qui lo stesso istante appare diversamente in UTC e a Roma. Il timestamp senza fuso mantiene la stessa rappresentazione.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def zone_example():\n    return spark.sql('''\n        SELECT\n          TIMESTAMP '2026-07-01 23:30:00Z' AS instant,\n          TIMESTAMP_NTZ '2026-07-01 23:30:00' AS wall_clock\n    ''')\n\ntry:\n    for zone in (\"UTC\", \"Europe/Rome\"):\n        spark.conf.set(\"spark.sql.session.timeZone\", zone)\n        print(zone)\n        example = zone_example()\n        example.select(\n            F.col(\"instant\").cast(\"string\"),\n            F.col(\"wall_clock\").cast(\"string\"),\n            F.window(\"instant\", \"1 day\").getField(\"start\").cast(\"string\").alias(\"window_start\"),\n        ).show(truncate=False)\nfinally:\n    spark.conf.set(\"spark.sql.session.timeZone\", \"UTC\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 23. Convertire da UTC a ora locale\n\nQui la sessione è in UTC. `from_utc_timestamp()` sposta il valore del timestamp così che, formattato in UTC, mostri l’ora locale di destinazione. Usa questa rappresentazione per visualizzare l’ora locale o estrarne la data e conserva l’istante originale per i confronti.\n\n## 24. Cluster con fusi diversi\n\nConfigura la sessione SQL esplicitamente e interpreta input con uno scarto orario quando un evento rappresenta un istante. Evita di dipendere dalle impostazioni del sistema operativo in Python o nella JVM. Nei passaggi con Python, usa conversioni esplicite con fuso o confronta i valori rispetto all’epoca per verificare che l’istante sia lo stesso.\n\n## 25. Preparare una colonna di date\n\nConserva il testo originale, usa un formato documentato, converti in modo sicuro e segnala i valori malformati separatamente da quelli mancanti. Per queste stringhe che contengono solo date e sono interpretate in UTC, converti il timestamp ottenuto in DATE e verifica il tipo del risultato e il numero di errori.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "rome = zone_example().select(\n    F.date_format(F.from_utc_timestamp(\"instant\", \"Europe/Rome\"), \"yyyy-MM-dd HH:mm\").alias(\"rome_label\")\n)\nassert rome.first().rome_label == \"2026-07-02 01:30\"\n\nincoming = spark.createDataFrame(\n    [(\"2026-11-04\",), (\"2026-02-30\",), (None,)],\n    \"raw_date STRING\",\n)\ntyped = incoming.withColumn(\n    \"order_date\", F.try_to_timestamp(\"raw_date\", F.lit(day_pattern)).cast(\"date\")\n)\nchecked = typed.withColumn(\"invalid_date\", F.col(\"raw_date\").isNotNull() & F.col(\"order_date\").isNull())\nassert checked.schema[\"order_date\"].dataType.simpleString() == \"date\"\nassert checked.filter(\"invalid_date\").count() == 1\nchecked.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Come distingui il mese dai minuti nel formato di una stringa data e ora?\n2. Perché un evento può comparire due volte nell’output delle finestre scorrevoli?\n3. Quale colonna dovresti conservare quando converti un istante in una rappresentazione locale?\n\n### Risposte\n\n1. Usa `MM` per il mese e `mm` per i minuti.\n2. Le finestre si sovrappongono e più di una può contenere l’evento.\n3. Conserva l’istante originale e ricava la rappresentazione locale in un’altra colonna.",
      "id": "cell-17"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}