08 · Lavorare con date e ore

Interpreta 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.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[2]").appName("08-dates-and-times").getOrCreate()
spark.conf.set("spark.sql.session.timeZone", "UTC")
spark.conf.set("spark.sql.timestampType", "TIMESTAMP_LTZ")
spark.conf.set("spark.sql.ansi.enabled", "true")
spark.conf.set("spark.sql.legacy.timeParserPolicy", "CORRECTED")

raw = spark.createDataFrame(
    [(101, "2026-11-01", "2026-11-01 09:05"),
     (102, "2026-11-02", "2026-11-02 12:30"),
     (103, None, "bad input")],
    "order_id INT, day_text STRING, time_text STRING",
)

1. Tipi date e timestamp

DATE 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.

2. Convertire testo in timestamp

Usa to_timestamp() con un formato di input quando sai che il testo è valido.

3. Convertire testo in data

Usa to_date() quando serve solo il giorno di calendario. Un input mancante resta null.

4. Formato di una data

In yyyy-MM-dd i campi sono anno, mese e giorno. Il formato deve descrivere il testo di input.

5. MM e mm

I formati distinguono maiuscole e minuscole. MM indica il mese, mm i minuti e HH l’ora su 24 ore.

6. Conversione sicura in timestamp

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.

day_pattern = "yyyy-MM-dd"
time_pattern = "yyyy-MM-dd HH:mm"
raw.filter("order_id < 103").select(F.to_timestamp("time_text", time_pattern)).show()
parsed = raw.select(
    "*",
    F.to_date("day_text", day_pattern).alias("order_day"),
    F.try_to_timestamp("time_text", F.lit(time_pattern)).alias("order_time"),
)
assert parsed.filter(F.col("order_time").isNull()).count() == 1
parsed.printSchema()

7. Estrarre parti di una data

Usa funzioni come year(), month() e dayofmonth() per ricavare campi numerici.

8. Formattare una data come testo

date_format() produce una stringa da visualizzare o esportare. Conserva la colonna tipizzata per i calcoli.

9. YYYY e yyyy

Usa 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.

parsed.select(
    "order_day", F.year("order_day").alias("year"), F.month("order_day").alias("month"),
    F.dayofmonth("order_day").alias("day"), F.date_format("order_day", "dd/MM/yyyy").alias("label"),
).show()

10. Troncare una data

Il troncamento porta un valore all’inizio di un’unità. Troncare al mese dà il primo giorno del mese.

11. dayofweek parte dalla domenica

dayofweek() in Spark restituisce 1 per domenica e 7 per sabato.

12. date_trunc dà un timestamp, trunc una data

Usa trunc(date, unit) per un DATE e date_trunc(unit, timestamp) per un timestamp. L’ordine degli argomenti cambia.

buckets = parsed.select(
    "order_id", F.trunc("order_day", "month").alias("month_date"),
    F.dayofweek("order_day").alias("weekday"),
    F.date_trunc("month", "order_time").alias("month_timestamp"),
)
buckets.printSchema()
assert buckets.filter("order_id = 101").first().weekday == 1

13. Differenze tra date

datediff(end, start) restituisce giorni di calendario con segno. Opera sulle date e non misura le ore trascorse.

14. Spostare una data

Usa date_add() e date_sub() per spostarti di giorni di calendario.

15. Creare un calendario

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.

16. add_months a fine mese

Quando il mese di destinazione non contiene il giorno di partenza, add_months() usa il suo ultimo giorno valido.

parsed.select(
    F.datediff("order_day", F.lit("2026-11-01")).alias("days_since_start"),
    F.date_add("order_day", 2).alias("follow_up"),
    F.date_sub("order_day", 2).alias("two_days_before"),
).show()
calendar = spark.range(1).select(
    F.explode(F.expr("sequence(DATE '2026-11-01', DATE '2026-11-07', INTERVAL 1 DAY)")).alias("day")
)
assert calendar.count() == 7
calendar.orderBy("day").show()
assert spark.sql("SELECT CAST(add_months(DATE '2024-01-31', 1) AS STRING) AS day").first().day == "2024-02-29"

17. Finestra temporale fissa

Una 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.

18. Finestra temporale scorrevole

Una finestra di dieci minuti con uno scorrimento di cinque minuti si sovrappone alle finestre vicine. Uno stesso evento può comparire in più finestre.

19. Sessione di eventi

Una 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.

events = spark.createDataFrame(
    [("A", "2026-11-01 09:02"), ("A", "2026-11-01 09:06"),
     ("A", "2026-11-01 09:10"), ("A", "2026-11-01 09:25"),
     ("B", "2026-11-01 09:06")],
    "customer STRING, event_text STRING",
).withColumn("event_time", F.to_timestamp("event_text", time_pattern))

fixed = events.groupBy(F.window("event_time", "10 minutes").alias("bucket")).count()
fixed.orderBy("bucket.start").show(truncate=False)
assert sorted(row["count"] for row in fixed.collect()) == [1, 1, 3]

sliding = events.groupBy(F.window("event_time", "10 minutes", "5 minutes").alias("bucket")).count()
assert sliding.agg(F.sum("count")).first()[0] == 10
sliding.orderBy("bucket.start").show(truncate=False)

sessions = events.groupBy("customer", F.session_window("event_time", "5 minutes").alias("session")).count()
assert sorted((r.customer, r["count"]) for r in sessions.collect()) == [("A", 1), ("A", 3), ("B", 1)]
sessions.orderBy("customer", "session.start").show(truncate=False)

20. Fuso orario della sessione

Il 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.

21. Finestre allineate a UTC

Le 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.

22. TIMESTAMP e TIMESTAMP_NTZ

Con 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.

def zone_example():
    return spark.sql('''
        SELECT
          TIMESTAMP '2026-07-01 23:30:00Z' AS instant,
          TIMESTAMP_NTZ '2026-07-01 23:30:00' AS wall_clock
    ''')

try:
    for zone in ("UTC", "Europe/Rome"):
        spark.conf.set("spark.sql.session.timeZone", zone)
        print(zone)
        example = zone_example()
        example.select(
            F.col("instant").cast("string"),
            F.col("wall_clock").cast("string"),
            F.window("instant", "1 day").getField("start").cast("string").alias("window_start"),
        ).show(truncate=False)
finally:
    spark.conf.set("spark.sql.session.timeZone", "UTC")

23. Convertire da UTC a ora locale

Qui 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.

24. Cluster con fusi diversi

Configura 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.

25. Preparare una colonna di date

Conserva 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.

rome = zone_example().select(
    F.date_format(F.from_utc_timestamp("instant", "Europe/Rome"), "yyyy-MM-dd HH:mm").alias("rome_label")
)
assert rome.first().rome_label == "2026-07-02 01:30"

incoming = spark.createDataFrame(
    [("2026-11-04",), ("2026-02-30",), (None,)],
    "raw_date STRING",
)
typed = incoming.withColumn(
    "order_date", F.try_to_timestamp("raw_date", F.lit(day_pattern)).cast("date")
)
checked = typed.withColumn("invalid_date", F.col("raw_date").isNotNull() & F.col("order_date").isNull())
assert checked.schema["order_date"].dataType.simpleString() == "date"
assert checked.filter("invalid_date").count() == 1
checked.show()

Verifica la comprensione

1. Come distingui il mese dai minuti nel formato di una stringa data e ora?
2. Perché un evento può comparire due volte nell’output delle finestre scorrevoli?
3. Quale colonna dovresti conservare quando converti un istante in una rappresentazione locale?

Risposte

1. Usa MM per il mese e mm per i minuti.
2. Le finestre si sovrappongono e più di una può contenere l’evento.
3. Conserva l’istante originale e ricava la rappresentazione locale in un’altra colonna.