Inserire dati con Spark Structured Streaming

L'inserimento dati in tempo reale consente di elaborare gli eventi man mano che si verificano anziché attendere intervalli di batch. Quando la tua organizzazione deve acquisire le letture dei sensori IoT, i dati clickstream o le transazioni finanziarie in tempo reale, Spark Structured Streaming offre le basi per l'inserimento continuo dei dati nel lakehouse.

In questa unità apprenderai come configurare job di Structured Streaming che leggono da origini di streaming e scrivono nelle tabelle di Unity Catalog.

Informazioni sul modello Structured Streaming

Structured Streaming considera i dati in ingresso come una tabella illimitata che cresce continuamente man mano che arrivano nuovi dati. Si usa la stessa API dataframe già nota, ma invece di spark.read, si usa spark.readStream per creare un dataframe di streaming che elabora i dati in modo incrementale.

Si consideri uno scenario in cui i dati dei sensori arrivano continuamente dalle apparecchiature di produzione. Anziché pianificare processi batch orari per inserire questi dati, è possibile configurare un processo di streaming che elabora ogni lettura entro pochi secondi dall'arrivo. Il motore di streaming tiene traccia dei record elaborati usando checkpoint, quindi, se il processo viene riavviato, riprende da dove è stato interrotto senza rielaborare o perdere dati.

Il modello fondamentale per l'inserimento in streaming prevede tre passaggi:

1. Configurare l'origine usando spark.readStream con i dettagli della connessione
2. Applicare trasformazioni per modellare i dati in base alle esigenze
3. Scrivere in una destinazione usando writeStream con una posizione del checkpoint

Connettersi a Kafka e a Hub eventi

Hub eventi di Azure fornisce un endpoint compatibile con Apache Kafka, il che significa che è possibile usare lo stesso connettore Kafka per entrambi i servizi. Questo approccio unificato semplifica il codice quando si utilizzano bus di messaggio diversi.

L'esempio seguente legge i messaggi da un topic Kafka:

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker-server:9092")
    .option("subscribe", "sensor-readings")
    .option("startingOffsets", "latest")
    .load())

Per Hub eventi di Azure, ci si connette tramite l'interfaccia compatibile con Kafka. La differenza principale è la configurazione di autenticazione, che usa SASL con la stringa di connessione di Hub eventi:

eh_namespace = "your-namespace"
eh_name = "your-eventhub"
connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

kafka_options = {
    "kafka.bootstrap.servers": f"{eh_namespace}.servicebus.windows.net:9093",
    "subscribe": eh_name,
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{connection_string}";'
}

df_stream = (spark.readStream
    .format("kafka")
    .options(**kafka_options)
    .load())

Quando si legge da Kafka o Event Hubs tramite l'interfaccia compatibile con Kafka, si riceve uno schema standard con le colonne key, value, topic, partition, offset e timestamp. Le key colonne e value contengono dati binari, quindi in genere è possibile eseguirne il cast in stringa e analizzare il contenuto:

from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("device_id", StringType()) \
    .add("temperature", DoubleType()) \
    .add("timestamp", StringType())

parsed_df = (df_stream
    .select(col("value").cast("string").alias("json_value"))
    .select(from_json(col("json_value"), schema).alias("data"))
    .select("data.*"))

Usare il connettore nativo di Event Hubs

Azure Databricks offre anche un connettore nativo di Hub eventi che semplifica l'autenticazione e fornisce l'accesso diretto alle funzionalità specifiche di Hub eventi. Questo connettore usa il eventhubs formato :

connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .load())

Il connettore nativo di Hub eventi restituisce uno schema diverso rispetto all'interfaccia compatibile con Kafka. Il payload dell'evento viene archiviato nella colonna body (anziché value), insieme ad altre colonne di metadati specifiche di Event Hubs.

# Extract and parse the event payload
payload = events.selectExpr("CAST(body AS STRING) AS payload")

È possibile configurare opzioni aggiuntive di Hub eventi, ad esempio il gruppo di consumer e la posizione iniziale:

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .option("eventhubs.consumerGroup", "$Default")
    .option("eventhubs.startingPosition", '{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}')
    .load())

L'opzione startingPosition controlla dove iniziare a leggere gli eventi. Usare -1 per gli eventi più recenti o configurare un offset, un numero di sequenza o un tempo di accodamento specifico per le esigenze di inserimento.

Configurare i percorsi dei checkpoint

I checkpoint sono il meccanismo che rende Structured Streaming a prova di errore. La directory del checkpoint archivia le informazioni sull'offset che tiene esattamente traccia di quali record sono stati elaborati. Se il processo di streaming si arresta in modo imprevisto, può riavviarlo e continuare dall'ultimo checkpoint di cui è stato eseguito il commit.

Quando inizi una scrittura di streaming, specifica il percorso del checkpoint:

checkpoint_path = "/checkpoints/sensor-ingestion"

query = (parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .toTable("catalog.schema.sensor_readings"))

Archiviare i checkpoint in una posizione affidabile e durevole, ad esempio Azure Data Lake Storage. I file di checkpoint sono di piccole dimensioni ma critici per il ripristino, quindi configura il percorso di archiviazione con criteri di backup e di conservazione appropriati.

Scrivere dati di streaming in tabelle di Unity Catalog

La scrittura nelle tabelle del catalogo Unity usa il formato Delta Lake, che supporta sia le operazioni di scrittura in streaming che le transazioni ACID. L'API writeStream esegue il mirroring dell'API batch write , ma include opzioni per l'elaborazione continua:

(parsed_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("bronze.iot.sensor_readings"))

La modalità di output determina la modalità di scrittura dei risultati:

• append: vengono aggiunte nuove righe alla tabella. Usare questa modalità per la maggior parte dei carichi di lavoro di ingestione.
• update: le righe modificate vengono aggiornate. Usare con le aggregazioni che producono aggiornamenti.
• complete: tutte le righe vengono riscritte. Usare con le aggregazioni che richiedono un ricalcolo completo.

Per gli scenari di inserimento in cui si aggiungono dati non elaborati, la modalità predefinita append è appropriata.

Controllare l'elaborazione con i trigger

I trigger determinano quando e con quale frequenza il motore di streaming elabora i dati. Azure Databricks supporta diverse opzioni di trigger:

Per l'inserimento continuo che viene eseguito a tempo indeterminato, omettere l'opzione trigger o usare un intervallo di tempo di elaborazione. Per i modelli batch incrementali in cui si vogliono elaborare tutti i dati disponibili in base a una pianificazione, usare availableNow:

(parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .trigger(availableNow=True)
    .toTable("catalog.schema.table_name"))

Il availableNow trigger elabora tutti i dati non elaborati dall'ultimo checkpoint, quindi termina. È possibile pianificare questo modello usando Processi di Databricks per l'esecuzione a intervalli regolari, combinando l'efficienza del flusso (solo elaborando nuovi dati) con la prevedibilità dei processi batch pianificati.

Monitorare le query di streaming

Le query di streaming attive vengono visualizzate nell'interfaccia utente di Spark nella scheda Structured Streaming . È anche possibile accedere allo stato delle query a livello di codice:

query = parsed_df.writeStream.format("delta").option("checkpointLocation", checkpoint_path).toTable("my_table")

# Check if the query is still running
print(query.isActive)

# Get the current status
print(query.status)

# View recent progress
print(query.recentProgress)

Usare il metodo awaitTermination() per bloccare l'esecuzione fino al completamento o in caso di errore del flusso. Per i carichi di lavoro di produzione, implementare la gestione degli errori per acquisire e rispondere agli errori di streaming:

try:
    query.awaitTermination()
except Exception as e:
    print(f"Stream failed: {e}")

Negli ambienti di produzione configurare il monitoraggio e gli avvisi per rilevare quando i processi di streaming sono in ritardo o riscontrano errori. Le metriche di streaming includono informazioni sul ritardo che consentono di comprendere se l'elaborazione mantiene il ritmo con i dati in ingresso.