24 · Elaborare flussi di dati

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

1. Soglia di ritardo

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

2. Leggere un flusso

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

from pathlib import Path
from tempfile import TemporaryDirectory

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

spark = SparkSession.builder.master("local[2]").appName("24-structured-streaming").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
spark.conf.set("spark.sql.session.timeZone", "UTC")
scratch = TemporaryDirectory(prefix="nb24-")
root = Path(scratch.name)
incoming = root / "incoming"
incoming.mkdir()
output = str(root / "output")
checkpoint = str(root / "checkpoint")
schema = "event_id STRING, event_time TIMESTAMP"

def publish(name, contents):
    staged = root / name
    staged.write_text(contents, encoding="utf-8")
    staged.rename(incoming / name)

publish("first.json", '''
{"event_id":"A","event_time":"2026-11-01T09:00:00Z"}
{"event_id":"A","event_time":"2026-11-01T09:01:00Z"}
{"event_id":"B","event_time":"2026-11-01T09:02:00Z"}
'''.strip() + "\n")
events = spark.readStream.schema(schema).option("maxFilesPerTrigger", 1).json(str(incoming))

3. Scrivere un flusso

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

4. Eliminare i duplicati in un flusso

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.

5. dropDuplicatesWithinWatermark su un DataFrame batch

L’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.

def run_available(frame):
    query = (
        frame.writeStream.format("parquet").outputMode("append")
        .option("checkpointLocation", checkpoint)
        .trigger(availableNow=True).start(output)
    )
    try:
        if not query.awaitTermination(120):
            raise TimeoutError("Streaming query did not finish within 120 seconds.")
        return query.recentProgress
    finally:
        if query.isActive:
            query.stop()

deduplicated = events.withWatermark("event_time", "10 minutes").dropDuplicatesWithinWatermark(["event_id"])
static_events = spark.read.schema(schema).json(str(incoming))
try:
    static_events.dropDuplicatesWithinWatermark(["event_id"]).collect()
except AnalysisException as error:
    print("Expected batch rejection:", error.getCondition())
else:
    raise AssertionError("The streaming deduplication operator should reject batch input.")
assert static_events.dropDuplicates(["event_id"]).count() == 2

6. Checkpoint di un flusso

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

7. Structured Streaming

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

8. Micro-batch

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

first_progress = run_available(deduplicated)
assert sorted(r.event_id for r in spark.read.parquet(output).collect()) == ["A", "B"]

publish("second.json", '''
{"event_id":"A","event_time":"2026-11-01T09:03:00Z"}
{"event_id":"C","event_time":"2026-11-01T09:04:00Z"}
'''.strip() + "\n")
second_progress = run_available(deduplicated)
assert sorted(r.event_id for r in spark.read.parquet(output).collect()) == ["A", "B", "C"]
for entry in first_progress + second_progress:
    print("Batch", entry["batchId"], "input rows", entry["numInputRows"])

9. count() su un flusso

L’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.

10. Controllare se un DataFrame è un flusso

Usa isStreaming sulla sorgente o su un DataFrame derivato. Una lettura batch della stessa directory restituisce false.

11. Modalità di uscita

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

12. Attivazione a intervallo

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.

from uuid import uuid4

try:
    events.count()
except AnalysisException as error:
    print("Expected streaming action rejection:", error.getCondition())
else:
    raise AssertionError("A batch count should reject streaming input.")
assert events.isStreaming and deduplicated.isStreaming
assert not static_events.isStreaming

def periodic_writer(frame):
    return frame.writeStream.trigger(processingTime="5 seconds")

table = "nb24_counts_" + uuid4().hex[:8]
running = events.agg(F.count("*").alias("event_rows"))
query = (
    running.writeStream.format("memory").queryName(table)
    .outputMode("complete").trigger(availableNow=True).start()
)
try:
    if not query.awaitTermination(120):
        raise TimeoutError("Count query did not finish within 120 seconds.")
    assert spark.table(table).first().event_rows == 5
finally:
    if query.isActive:
        query.stop()
    spark.catalog.dropTempView(table)
    scratch.cleanup()

Verifica la comprensione

1. Quale orologio governa la watermark?
2. Perché A appare una sola volta dopo il riavvio?
3. Qual è la differenza tra trigger e modalità di uscita?

Risposte

1. Il tempo degli eventi osservato.
2. La query riavviata recupera lo stato di deduplicazione dal checkpoint e l’ID ripetuto rientra ancora nell’orizzonte di conservazione dello stato.
3. Il trigger pianifica il lavoro. La modalità di uscita determina quali righe del risultato vengono emesse.