Führen Sie Ihrem ersten strukturierten Streaming-Workload aus

Dieser Artikel enthält Codebeispiele und Erläuterungen grundlegender Konzepte, die zum Ausführen Ihrer ersten strukturierten Streaming-Abfragen auf Azure Databricks erforderlich sind. Sie können strukturiertes Streaming für nahezu Echtzeit- und inkrementelle Verarbeitungsworkloads verwenden.

Strukturiertes Streaming ist eine von mehreren Technologien, die Streamingtabellen in Delta Live-Tabellen nutzen. Databricks empfiehlt die Verwendung von Delta Live Tables für alle neuen ETL-, Aufnahme- und strukturierten Streaming-Workloads. Weitere Informationen finden Sie unter Was sind Delta Live-Tabellen?.

Verwenden des automatischen Ladens zum Lesen von Streamingdaten aus dem Objektspeicher

Im folgenden Beispiel wird das Laden von JSON-Daten mit Auto Loader veranschaulicht, das zum Kennzeichnen von Format und Optionen cloudFiles verwendet. Die schemaLocation-Option ermöglicht Schemarückschluss und -evolution. Fügen Sie den folgenden Code in eine Zelle des Databricks-Notebooks ein und führen Sie die Zelle aus, um einen Streaming-DataFrame mit dem Namen raw_df zu erstellen:

file_path = "/databricks-datasets/structured-streaming/events"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

raw_df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint_path)
    .load(file_path)
)

Wie bei anderen Lesevorgängen in Azure Databricks werden beim Konfigurieren von Streaminglesedaten keine Daten geladen. Sie müssen eine Aktion für die Daten auslösen, bevor der Datenstrom beginnt.

Durchführen einer Streamingtransformation

Strukturiertes Streaming unterstützt die meisten Transformationen, die in Azure Databricks und Spark SQL verfügbar sind. Sie können sogar MLflow-Modelle als UDFs laden und Streamingvorhersagen als Transformation erstellen.

Im folgenden Codebeispiel wird eine einfache Transformation abgeschlossen, um die aufgenommenen JSON-Daten mit zusätzlichen Informationen mithilfe von Spark SQL-Funktionen zu erweitern:

from pyspark.sql.functions import col, current_timestamp

transformed_df = (raw_df.select(
    "*",
    col("_metadata.file_path").alias("source_file"),
    current_timestamp().alias("processing_time")
    )
)

Das resultierende transformed_df enthält Abfrageanweisungen zum Laden und Transformieren jedes Datensatzes, sobald er in der Datenquelle eingeht.

Durchführen eines inkrementellen Batchschreibvorgangs in Delta Lake

Im folgenden Beispiel wird mithilfe eines angegebenen Dateipfads und Prüfpunkts in Delta Lake geschrieben.

Die availableNow-Einstellung für den Trigger weist strukturiertes Streaming an, alle zuvor unverarbeiteten Datensätze aus dem Quelldatensatz zu verarbeiten und dann herunterzufahren, sodass Sie den folgenden Code sicher ausführen können, ohne sich Gedanken über das Verlassen eines Datenstroms machen zu müssen:

target_path = "/tmp/ss-tutorial/"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

transformed_df.writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .option("path", target_path)
    .start()

In diesem Beispiel werden keine neuen Datensätze in unserer Datenquelle eintreffen, sodass die Ausführung dieses Codes keine neuen Datensätze erfasst.

Lesen von Daten aus Delta Lake, Transformation und Schreiben in Delta Lake

Delta Lake verfügt über umfangreiche Unterstützung für die Arbeit mit strukturiertem Streaming sowohl als Quelle als auch als Senke. Weitere Informationen finden Sie unter Delta-Tabelle: Streaming für Lese- und Schreibvorgänge.

Das folgende Beispiel zeigt eine Beispielsyntax, um alle neuen Datensätze aus einer Delta-Tabelle inkrementell zu laden, sie mit einer Momentaufnahme einer anderen Delta-Tabelle zu verbinden und in eine Delta-Tabelle zu schreiben:

(spark.readStream
    .table("<table-name1>")
    .join(spark.read.table("<table-name2>"), on="<id>", how="left")
    .writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", "<checkpoint-path>")
    .toTable("<table-name3>")
)

Sie müssen über die erforderlichen Berechtigungen verfügen, um Quelltabellen zu lesen und in Zieltabellen und den angegebenen Prüfpunktspeicherort zu schreiben. Füllen Sie alle Parameter aus, die mit gewinkelten Klammern (<>) gekennzeichnet sind, indem Sie die relevanten Werte für Ihre Datenquellen und Senken verwenden.

Lesen von Daten aus Kafka, Transformieren und Schreiben in Kafka

Apache Kafka und andere Nachrichtenbusse bieten mit die niedrigste Latenz, die für große Datasets verfügbar ist. Sie können Azure Databricks verwenden, um Transformationen auf Daten anzuwenden, die aus Kafka aufgenommen wurden, und dann Daten zurück in Kafka schreiben.

Das folgende Codebeispiel veranschaulicht ein einfaches Muster zum Anreichern von Daten aus Kafka durch Verknüpfen mit Daten in einer Delta-Tabelle und anschließendes Zurückschreiben in Kafka:

(spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("subscribe", "<topic>")
    .option("startingOffsets", "latest")
    .load()
    .join(spark.read.table("<table-name>"), on="<id>", how="left")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("topic", "<topic>")
    .option("checkpointLocation", "<checkpoint-path>")
    .start()
)

Sie müssen über die erforderlichen Berechtigungen für den Zugriff auf Ihren Kafka-Dienst verfügen. Füllen Sie alle Parameter aus, die mit gewinkelten Klammern (<>) gekennzeichnet sind, indem Sie die relevanten Werte für Ihre Datenquellen und Senken verwenden. Siehe Stream-Verarbeitung mit Apache Kafka und Azure Databricks.