{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 24 · Elaborare flussi di dati\n\nLeggi file di eventi in modo incrementale, deduplica per ID evento e riprendi da un checkpoint. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"`, poi usa Run All. Input e output sono temporanei.",
      "id": "cell-01"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Soglia di ritardo\n\nUna watermark permette a Spark di limitare lo stato usando il tempo degli eventi. Con un ritardo di dieci minuti, Spark osserva gli orari degli eventi per far avanzare una soglia. Il ritardo è relativo al tempo degli eventi osservato, anziché all’orologio di sistema. I record entro il ritardo configurato sono protetti dallo scarto per ritardo. Quelli più vecchi possono essere elaborati o scartati, secondo la watermark effettiva e l’operatore.\n\n## 2. Leggere un flusso\n\nUsa `readStream` con uno schema esplicito per questa sorgente di file JSON. Colloca file completi nella directory di input e assegna nomi univoci ai file nuovi.",
      "id": "cell-02"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nfrom pyspark.errors import AnalysisException\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"24-structured-streaming\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nspark.conf.set(\"spark.sql.session.timeZone\", \"UTC\")\nscratch = TemporaryDirectory(prefix=\"nb24-\")\nroot = Path(scratch.name)\nincoming = root / \"incoming\"\nincoming.mkdir()\noutput = str(root / \"output\")\ncheckpoint = str(root / \"checkpoint\")\nschema = \"event_id STRING, event_time TIMESTAMP\"\n\ndef publish(name, contents):\n    staged = root / name\n    staged.write_text(contents, encoding=\"utf-8\")\n    staged.rename(incoming / name)\n\npublish(\"first.json\", '''\n{\"event_id\":\"A\",\"event_time\":\"2026-11-01T09:00:00Z\"}\n{\"event_id\":\"A\",\"event_time\":\"2026-11-01T09:01:00Z\"}\n{\"event_id\":\"B\",\"event_time\":\"2026-11-01T09:02:00Z\"}\n'''.strip() + \"\\n\")\nevents = spark.readStream.schema(schema).option(\"maxFilesPerTrigger\", 1).json(str(incoming))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-03"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 3. Scrivere un flusso\n\nIl writer del flusso specifica modalità di uscita, destinazione, checkpoint e trigger. L’avvio restituisce un oggetto StreamingQuery. La funzione sotto usa una destinazione Parquet in append e attende che siano elaborati i file disponibili.\n\n## 4. Eliminare i duplicati in un flusso\n\n`dropDuplicatesWithinWatermark([\"event_id\"])` mantiene stato tra batch e usa una watermark per limitare il tempo in cui conserva gli ID. Scegli un ritardo maggiore della differenza temporale attesa tra copie dello stesso evento. Qui due copie di A sono distanti un minuto.\n\n## 5. dropDuplicatesWithinWatermark su un DataFrame batch\n\nL’operatore richiede un DataFrame in streaming. Usa `dropDuplicates()` per un batch quando serve solo una riga per chiave e un ordinamento esplicito quando importa quale riga sopravvive.",
      "id": "cell-04"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def run_available(frame):\n    query = (\n        frame.writeStream.format(\"parquet\").outputMode(\"append\")\n        .option(\"checkpointLocation\", checkpoint)\n        .trigger(availableNow=True).start(output)\n    )\n    try:\n        if not query.awaitTermination(120):\n            raise TimeoutError(\"Streaming query did not finish within 120 seconds.\")\n        return query.recentProgress\n    finally:\n        if query.isActive:\n            query.stop()\n\ndeduplicated = events.withWatermark(\"event_time\", \"10 minutes\").dropDuplicatesWithinWatermark([\"event_id\"])\nstatic_events = spark.read.schema(schema).json(str(incoming))\ntry:\n    static_events.dropDuplicatesWithinWatermark([\"event_id\"]).collect()\nexcept AnalysisException as error:\n    print(\"Expected batch rejection:\", error.getCondition())\nelse:\n    raise AssertionError(\"The streaming deduplication operator should reject batch input.\")\nassert static_events.dropDuplicates([\"event_id\"]).count() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-05"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Checkpoint di un flusso\n\nIl checkpoint registra offset elaborati e stato per consentire a un riavvio compatibile di riprendere. Tienilo separato da input e output. Riutilizzalo solo per la stessa query e destinazione compatibili.\n\n## 7. Structured Streaming\n\nIl DataFrame in streaming descrive una tabella senza limite. Spark applica la query in modo incrementale man mano che arrivano gli input. Available Now elabora i dati disponibili e poi si ferma, rendendo finito l’esempio.\n\n## 8. Micro-batch\n\nUn micro-batch elabora una parte dell’input disponibile. La query può anche riportare batch con zero righe in ingresso. Leggi le informazioni di avanzamento per capire quali batch contengono dati.",
      "id": "cell-06"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "first_progress = run_available(deduplicated)\nassert sorted(r.event_id for r in spark.read.parquet(output).collect()) == [\"A\", \"B\"]\n\npublish(\"second.json\", '''\n{\"event_id\":\"A\",\"event_time\":\"2026-11-01T09:03:00Z\"}\n{\"event_id\":\"C\",\"event_time\":\"2026-11-01T09:04:00Z\"}\n'''.strip() + \"\\n\")\nsecond_progress = run_available(deduplicated)\nassert sorted(r.event_id for r in spark.read.parquet(output).collect()) == [\"A\", \"B\", \"C\"]\nfor entry in first_progress + second_progress:\n    print(\"Batch\", entry[\"batchId\"], \"input rows\", entry[\"numInputRows\"])",
      "execution_count": null,
      "outputs": [],
      "id": "cell-07"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. count() su un flusso\n\nL’azione batch `count()` non può eseguire direttamente un input senza limite. Un’aggregazione in streaming può invece definire un conteggio progressivo e pubblicarlo in una destinazione compatibile.\n\n## 10. Controllare se un DataFrame è un flusso\n\nUsa `isStreaming` sulla sorgente o su un DataFrame derivato. Una lettura batch della stessa directory restituisce false.\n\n## 11. Modalità di uscita\n\nAppend emette nuove righe di output. Update emette le righe cambiate dal trigger precedente. Complete emette l’intera tabella dei risultati. Operatori e destinazioni determinano le combinazioni valide. Qui Parquet usa append. Una destinazione in memoria con modalità complete può mostrare un piccolo conteggio progressivo.\n\n## 12. Attivazione a intervallo\n\n`trigger(processingTime=\"5 seconds\")` richiede micro-batch periodici. L’intervallo pianifica il lavoro, senza garantire che ogni batch finisca entro quel tempo. Il notebook finito usa Available Now. La funzione mostra come configurare un writer periodico senza avviare una query di lunga durata.",
      "id": "cell-08"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from uuid import uuid4\n\ntry:\n    events.count()\nexcept AnalysisException as error:\n    print(\"Expected streaming action rejection:\", error.getCondition())\nelse:\n    raise AssertionError(\"A batch count should reject streaming input.\")\nassert events.isStreaming and deduplicated.isStreaming\nassert not static_events.isStreaming\n\ndef periodic_writer(frame):\n    return frame.writeStream.trigger(processingTime=\"5 seconds\")\n\ntable = \"nb24_counts_\" + uuid4().hex[:8]\nrunning = events.agg(F.count(\"*\").alias(\"event_rows\"))\nquery = (\n    running.writeStream.format(\"memory\").queryName(table)\n    .outputMode(\"complete\").trigger(availableNow=True).start()\n)\ntry:\n    if not query.awaitTermination(120):\n        raise TimeoutError(\"Count query did not finish within 120 seconds.\")\n    assert spark.table(table).first().event_rows == 5\nfinally:\n    if query.isActive:\n        query.stop()\n    spark.catalog.dropTempView(table)\n    scratch.cleanup()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-09"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale orologio governa la watermark?\n2. Perché A appare una sola volta dopo il riavvio?\n3. Qual è la differenza tra trigger e modalità di uscita?\n\n### Risposte\n\n1. Il tempo degli eventi osservato.\n2. La query riavviata recupera lo stato di deduplicazione dal checkpoint e l’ID ripetuto rientra ancora nell’orizzonte di conservazione dello stato.\n3. Il trigger pianifica il lavoro. La modalità di uscita determina quali righe del risultato vengono emesse.",
      "id": "cell-10"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}