Erfassen von Daten mit Spark Structured Streaming

Die Erfassung von Echtzeitdaten ermöglicht es Ihnen, Ereignisse so zu verarbeiten, wie sie auftreten, anstatt auf Batchintervalle zu warten. Wenn Ihre Organisation IoT-Sensorwerte, Clickstream-Daten oder Finanztransaktionen erfassen muss, bietet Spark Structured Streaming die Grundlage für die kontinuierliche Datenaufnahme in Ihr Lakehouse.

In dieser Lektion erfahren Sie, wie Sie strukturierte Streamingaufträge konfigurieren, die aus Streamingquellen lesen und in Unity Catalog-Tabellen schreiben.

Grundlegendes zum Strukturierten Streaming-Modell

Strukturiertes Streaming behandelt eingehende Daten als eine ungebundene Tabelle, die kontinuierlich wächst, wenn neue Daten eingehen. Sie verwenden dieselbe DataFrame-API, die Sie bereits kennen, aber anstelle von spark.read, verwenden Sie spark.readStream zum Erstellen eines Streaming-DataFrames, der Daten inkrementell verarbeitet.

Stellen Sie sich ein Szenario vor, in dem Sensordaten kontinuierlich von fertigungstechnischen Geräten empfangen werden. Anstatt stündliche Batchaufträge für die Erfassung dieser Daten zu planen, können Sie einen Streamingauftrag konfigurieren, der jeden Lesevorgang innerhalb von Sekunden nach seiner Ankunft verarbeitet. Die Streaming-Engine verfolgt, welche Datensätze mithilfe von Prüfpunkten verarbeitet wurden. Wenn der Auftrag neu gestartet wird, setzt es genau dort fort, wo es aufgehört hat, ohne dass Daten erneut verarbeitet werden oder verloren gehen.

Das grundlegende Muster für die Streamingaufnahme umfasst drei Schritte:

1. Konfigurieren der Quelle mithilfe von spark.readStream Verbindungsdetails
2. Anwenden von Transformationen zum Gestalten der Daten nach Bedarf
3. Schreiben auf eine Senke unter Verwendung von writeStream mit einem Kontrollpunkt

Herstellen einer Verbindung mit Kafka und Event Hubs

Azure Event Hubs bietet einen Apache Kafka-kompatiblen Endpunkt, was bedeutet, dass Sie denselben Kafka-Connector für beide Dienste verwenden können. Dieser einheitliche Ansatz vereinfacht Ihren Code beim Arbeiten mit verschiedenen Nachrichtenbussen.

Im folgenden Beispiel werden Nachrichten aus einem Kafka-Thema gelesen:

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

Für Azure Event Hubs stellen Sie eine Verbindung über die kafka-kompatible Schnittstelle her. Der Hauptunterschied ist die Authentifizierungskonfiguration, die SASL mit Ihrer Event Hubs-Verbindungszeichenfolge verwendet:

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())

Beim Lesen von Kafka oder Event Hubs über die kafka-kompatible Schnittstelle erhalten Sie ein Standardschema mit key, , value, topic, partition, offsetund timestamp Spalten. Die Spalten key und value enthalten Binärdaten, sodass Sie sie in der Regel in eine Zeichenfolge umwandeln und den Inhalt analysieren:

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.*"))

Verwenden Sie den nativen Event Hubs-Connector

Azure Databricks bietet auch einen nativen Event Hubs-Connector, der die Authentifizierung vereinfacht und direkten Zugriff auf Event Hubs-spezifische Features bietet. Dieser Connector verwendet das eventhubs Format:

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

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

Der native Event Hubs-Connector gibt ein anderes Schema als die kafka-kompatible Schnittstelle zurück. Die Ereignisnutzlast wird in der body Spalte (anstelle von value) zusammen mit zusätzlichen Event Hubs-spezifischen Metadatenspalten gespeichert:

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

Sie können zusätzliche Event Hubs-Optionen wie Consumergruppe und Startposition konfigurieren:

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())

Die startingPosition Option steuert, wo mit dem Lesen von Ereignissen begonnen werden soll. Verwenden Sie -1 für die neuesten Ereignisse oder konfigurieren Sie einen bestimmten Offset, eine Sequenznummer oder eine Warteschlangenzeit für Ihre Erfassungsanforderungen.

Konfigurieren von Kontrollpunkten

Prüfpunkte sind der Mechanismus, der strukturiertes Streaming fehlertolerant macht. Im Prüfpunktverzeichnis werden die Offsetinformationen gespeichert, mit denen genau nachverfolgt wird, welche Datensätze verarbeitet wurden. Wenn Ihr Streaming-Auftrag unerwartet anhält, kann er neu gestartet und ab dem letzten übertragenen Kontrollpunkt fortgesetzt werden.

Wenn Sie einen Streaming-Schreibvorgang starten, geben Sie den Ort des Kontrollpunkts an:

checkpoint_path = "/checkpoints/sensor-ingestion"

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

Speichern Sie Ihre Prüfpunkte an einem zuverlässigen, dauerhaften Ort wie Azure Data Lake Storage. Die Prüfpunktdateien sind klein, aber wichtig für die Wiederherstellung. Konfigurieren Sie daher Ihren Speicherort mit den entsprechenden Sicherungs- und Aufbewahrungsrichtlinien.

Schreiben von Streamingdaten in Unity-Katalogtabellen

Das Schreiben in Unity Catalog-Tabellen verwendet das Delta Lake-Format, das sowohl Streaming-Schreibvorgänge als auch ACID-Transaktionen unterstützt. Die writeStream API spiegelt die Batch-API write , enthält jedoch Optionen für die kontinuierliche Verarbeitung:

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

Der Ausgabemodus bestimmt, wie Ergebnisse geschrieben werden:

• append: Neue Zeilen werden der Tabelle hinzugefügt. Verwenden Sie diesen Modus für die meisten Erfassungsworkloads.
• update: Geänderte Zeilen werden aktualisiert. Wird mit Aggregationen verwendet, die Updates erzeugen.
• vollständig: Alle Zeilen werden umgeschrieben. Wird für Aggregationen verwendet, die eine vollständige Neuberechnung erfordern.

Für Aufnahmeszenarien, in denen Rohdaten angefügt werden, ist der Standardmodus append geeignet.

Steuerung der Verarbeitung durch Trigger

Trigger bestimmen, wann und wie häufig das Streamingmodul Daten verarbeitet. Azure Databricks unterstützt mehrere Triggeroptionen:

Für eine kontinuierliche Erfassung, die auf unbestimmte Zeit läuft, lassen Sie die Option Trigger weg oder verwenden Sie ein Zeitintervall für die Verarbeitung. Für inkrementelle Batchmuster, bei denen Sie alle verfügbaren Daten nach einem Zeitplan verarbeiten möchten, verwenden Sie availableNow.

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

Der availableNow Trigger verarbeitet alle unverarbeiteten Daten seit dem letzten Prüfpunkt und beendet dann. Dieses Muster kann mithilfe von Databricks-Aufträgen in regelmäßigen Intervallen geplant werden, wobei nur die neuen Daten verarbeitet werden und die Effizienz des Streamings mit der Planbarkeit geplanter Batchaufträge kombiniert wird.

Überwachen von Streamingabfragen

Aktive Streamingabfragen werden auf der Spark-Benutzeroberfläche auf der Registerkarte "Strukturiertes Streaming " angezeigt. Sie können auch programmgesteuert auf den Abfragestatus zugreifen:

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)

Verwenden Sie die Methode, um die awaitTermination() Ausführung zu blockieren, bis der Datenstrom abgeschlossen ist oder fehlschlägt. Implementieren Sie für Produktionsworkloads die Fehlerbehandlung, um Streamingfehler zu erfassen und darauf zu reagieren:

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

Konfigurieren Sie in Produktionsumgebungen die Überwachung und die Warnungen, um festzustellen, ob Streamingaufträge zurückfallen und Fehler auftreten. Streamingmetriken enthalten Verzögerungsinformationen, mit denen Sie verstehen können, ob Ihre Verarbeitung mit eingehenden Daten schritthält.