Flüsse in Lakeflow-Pipelines verwenden

Datenflüsse in einer Lakeflow-Pipeline übertragen Daten in eine Streamingtabelle oder eine materialisierte Ansicht. Die folgenden Beispiele zeigen, wie Sie Standard-Datenflüsse definieren, einen Datenfluss getrennt von seinem Ziel definieren, aus mehreren Kafka-Topics in eine Streaming-Tabelle schreiben, eine einmalige Rückbefüllung durchführen und UNION-Abfragen durch Append-Flow-Verarbeitung ersetzen.

Eine Übersicht über Flüsse finden Sie unter Laden und Verarbeiten von Daten inkrementell mit Lakeflow-Pipelineflüssen.

Beispiel: Erstellen eines Standardflusses

Wenn Sie eine Pipeline erstellen, definieren Sie in der Regel eine Tabelle oder Ansicht zusammen mit der Abfrage, die sie unterstützt. Diese Abfrage erstellt z. B. eine Streamingtabelle mit dem Namen customers_silver, indem sie aus customers_bronze liest. Die Streamingtabelle und der zugehörige Standardfluss werden in einem einzigen Schritt erstellt.

SQL

CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(customers_bronze)

Python

from pyspark import pipelines as dp

@dp.table()
def customers_silver():
  return spark.readStream.table("customers_bronze")

Der Standardfluss für eine Streamingtabelle ist ein Anfügefluss , der neue Zeilen mit jeder Aktualisierung hinzufügt und denselben Namen wie das Ziel hat. Dies ist die am häufigsten verwendete Methode zum Verwenden von Pipelines – Erstellen eines Flusses und seines Ziels in einem einzigen Schritt – und Sie können ihn zum Aufnehmen oder Transformieren von Daten verwenden. Weitere Informationen zu Flow-Konzepten finden Sie unter Daten mit Lakeflow-Pipeline-Flows inkrementell laden und verarbeiten.

Beispiel: Definieren eines Flusses getrennt von seinem Ziel

Sie können auch einen Fluss für eine Tabelle erstellen, die Sie separat definiert haben. Das Ergebnis ist identisch mit dem Erstellen eines Standardflusses, einschließlich der Verwendung desselben Namens für die Streamingtabelle und des Flusses:

Python

from pyspark import pipelines as dp

# create streaming table
dp.create_streaming_table("customers_silver")

# add a flow
@dp.append_flow(
  target = "customers_silver")
def customer_silver():
  return spark.readStream.table("customers_bronze")

SQL

-- create a streaming table
CREATE OR REFRESH STREAMING TABLE customers_silver;

-- add a flow
CREATE FLOW customers_silver
AS INSERT INTO customers_silver BY NAME
SELECT * FROM STREAM(customers_bronze);

Wenn Sie einen Fluss getrennt von seinem Ziel definieren, können Sie mehrere Flüsse erstellen, die Daten an dasselbe Ziel anfügen. Verwenden Sie den @dp.append_flow Dekorateur in der Python-Schnittstelle oder der CREATE FLOW...INSERT INTO Klausel in der SQL-Schnittstelle, um Flüsse für Aufgaben wie die folgenden hinzuzufügen:

Fügen Sie Streamingquellen hinzu, die Daten an eine vorhandene Streamingtabelle anfügen, ohne dass eine vollständige Aktualisierung erforderlich ist. Sie können beispielsweise eine Tabelle haben, die regionale Daten aus jeder Region kombiniert, in der Sie arbeiten. Wenn neue Regionen eingeführt werden, können Sie der Tabelle die neuen Regionsdaten hinzufügen, ohne eine vollständige Aktualisierung durchzuführen. Siehe Beispiel: Schreiben in eine Streamingtabelle aus mehreren Kafka-Themen.
Aktualisieren Sie eine Streamingtabelle, indem Sie fehlende historische Daten (Backfilling) anfügen. Sie können die INSERT INTO ONCE Syntax verwenden, um ein historisches Rückfüllen zu erstellen, das einmal ausgeführt wird. Siehe Beispiel: Einmalige Datennachfüllung ausführen und Historische Daten mit Pipelines nachfüllen.
Kombinieren Sie Daten aus mehreren Quellen, und schreiben Sie in eine einzelne Streamingtabelle, anstatt die UNION Klausel in einer Abfrage zu verwenden. Die Verwendung der Anfügeflowverarbeitung anstelle von UNION ermöglicht es Ihnen, die Zieltabelle inkrementell zu aktualisieren, ohne ein vollständiges Aktualisierungsupdateauszuführen. Siehe Beispiel: Verwenden der Anfügeflussverarbeitung anstelle von UNION.

Verwenden Sie für Python-Abfragen die Funktion create_streaming_table() zum Erstellen einer Zieltabelle.

Important

Wenn Sie Datenqualitätseinschränkungen mit Erwartungen definieren müssen, definieren Sie die Erwartungen an die Zieltabelle als Teil der create_streaming_table() Funktion oder einer vorhandenen Tabellendefinition. Sie können die Erwartungen in der @append_flow Definition nicht definieren.
Flüsse werden durch einen Flussnamen identifiziert, und dieser Name wird verwendet, um Streamingprüfpunkte zu identifizieren. Die Verwendung des Flussnamens zur Identifizierung des Prüfpunkts bedeutet Folgendes:
- Wenn ein vorhandener Workflow in einer Pipeline umbenannt wird, wird der Kontrollpunkt nicht übernommen, und der umbenannte Workflow gilt als ein völlig neuer Prozess.
- Sie können einen Flussnamen in einer Pipeline nicht wiederverwenden, da der vorhandene Prüfpunkt nicht mit der neuen Flussdefinition übereinstimmt.

Beispiel: Schreiben in eine Streamingtabelle aus mehreren Kafka-Themen

Im folgenden Beispiel wird eine Streamingtabelle mit dem Namen kafka_target erstellt, und es wird aus zwei Kafka-Themen in die Streamingtabelle geschrieben:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("kafka_target")

# Kafka stream from multiple topics
@dp.append_flow(target = "kafka_target")
def topic1():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,...")
      .option("subscribe", "topic1")
      .load()
  )

@dp.append_flow(target = "kafka_target")
def topic2():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,...")
      .option("subscribe", "topic2")
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_target;

CREATE FLOW
  topic1
AS INSERT INTO
  kafka_target BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic1');

CREATE FLOW
  topic2
AS INSERT INTO
  kafka_target BY NAME
SELECT * FROM
  read_kafka(bootstrapServers => 'host1:port1,...', subscribe => 'topic2');

Weitere Informationen zur read_kafka() in den SQL-Abfragen verwendeten Tabellenwertfunktion finden Sie unter read_kafka in der SQL-Sprachreferenz.

In Python können Sie programmgesteuert mehrere Flüsse erstellen, die auf eine einzelne Tabelle abzielen. Das folgende Beispiel zeigt dieses Muster für eine Liste der Kafka-Themen.

Hinweis

Dieses Muster hat die gleichen Anforderungen wie die Verwendung einer for Schleife zum Erstellen von Tabellen. Sie müssen einen Python-Wert explizit an die Funktion übergeben, die den Fluss definiert. Siehe Erstellen von Tabellen in einer for Schleife.

from pyspark import pipelines as dp

dp.create_streaming_table("kafka_target")

topic_list = ["topic1", "topic2", "topic3"]

for topic_name in topic_list:

  @dp.append_flow(target = "kafka_target", name=f"{topic_name}_flow")
  def topic_flow(topic=topic_name):
    return (
      spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "host1:port1,...")
        .option("subscribe", topic)
        .load()
    )

Beispiel: Ausführen eines einmaligen Datenrückfüllens

Wenn Sie eine Abfrage ausführen möchten, um Daten an eine vorhandene Streamingtabelle anzufügen, verwenden Sie append_flow.

Nach dem Anfügen einer Reihe vorhandener Daten haben Sie mehrere Optionen:

Wenn die Abfrage neue Daten anhängen soll, falls diese im Backfill-Verzeichnis eingehen, lassen Sie die Abfrage unverändert.
Wenn dies ein einmaliger Backfill sein soll und nie wieder ausgeführt werden soll, entfernen Sie die Abfrage, nachdem die Pipeline einmal ausgeführt wurde.
Wenn Sie möchten, dass die Abfrage einmal ausgeführt wird und nur dann erneut ausgeführt wird, wenn die Daten vollständig aktualisiert werden, legen Sie den once-Parameter auf True im Anfügefluss fest. Verwenden Sie INSERT INTO ONCEin SQL .

In den folgenden Beispielen wird eine Abfrage ausgeführt, um verlaufsgeschichtliche Daten an eine Streamingtabelle anzufügen:

Python

from pyspark import pipelines as dp

@dp.table()
def csv_target():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format","csv")
    .load("path/to/sourceDir")

@dp.append_flow(
  target = "csv_target",
  once = True)
def backfill():
  return spark.read
    .format("cloudFiles")
    .option("cloudFiles.format","csv")
    .load("path/to/backfill/data/dir")

SQL

CREATE OR REFRESH STREAMING TABLE csv_target
AS SELECT * FROM
  read_files(
    "path/to/sourceDir",
    "csv"
  );

CREATE FLOW
  backfill
AS INSERT INTO ONCE
  csv_target BY NAME
SELECT * FROM
  read_files(
    "path/to/backfill/data/dir",
    "csv"
  );

Ein ausführlicheres Beispiel finden Sie unter "Zurückfüllen von historischen Daten mit Pipelines".

Beispiel: Verwenden Sie die Anfügeflussverarbeitung anstelle von `UNION`

Anstatt eine Abfrage mit einer UNION Klausel zu verwenden, können Sie Anfügeflussabfragen verwenden, um mehrere Quellen zu kombinieren und in eine einzelne Streamingtabelle zu schreiben. Mithilfe von Anfügeflowabfragen anstelle von UNION können Sie an eine Streamingtabelle aus mehreren Quellen anfügen, ohne eine vollständige Aktualisierung auszuführen.

Das folgende Python-Beispiel enthält eine Abfrage, die mehrere Datenquellen mit einer UNION Klausel kombiniert:

@dp.create_table(name="raw_orders")
def unioned_raw_orders():
  raw_orders_us = (
    spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/path/to/orders/us")
  )

  raw_orders_eu = (
    spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/path/to/orders/eu")
  )

  return raw_orders_us.union(raw_orders_eu)

In den folgenden Beispielen wird die UNION-Abfrage durch Anfügeflowabfragen ersetzt:

Python

dp.create_streaming_table("raw_orders")

@dp.append_flow(target="raw_orders")
def raw_orders_us():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/us")

@dp.append_flow(target="raw_orders")
def raw_orders_eu():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/eu")

# Additional flows can be added without the full refresh that a UNION query would require:
@dp.append_flow(target="raw_orders")
def raw_orders_apac():
  return spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .load("/path/to/orders/apac")

SQL

CREATE OR REFRESH STREAMING TABLE raw_orders;

CREATE FLOW
  raw_orders_us
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/us",
    format => "csv"
  );

CREATE FLOW
  raw_orders_eu
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/eu",
    format => "csv"
  );

-- Additional flows can be added without the full refresh that a UNION query would require:
CREATE FLOW
  raw_orders_apac
AS INSERT INTO
  raw_orders BY NAME
SELECT * FROM
  STREAM read_files(
    "/path/to/orders/apac",
    format => "csv"
  );

Beispiel: Verwenden Sie `transformWithState` zum Überwachen von Sensor-Heartbeats

Das Beispiel zeigt einen zustandsbehafteten Prozessor, der aus Kafka liest und überprüft, ob Sensoren regelmäßig Herzschläge abgeben. Wenn ein Heartbeat nicht innerhalb von 5 Minuten empfangen wird, schreibt der Prozessor einen Eintrag in die Ziel-Delta-Tabelle zur Analyse.

Weitere Informationen zum Erstellen benutzerdefinierter zustandsbehafteter Anwendungen finden Sie unter Erstellen einer benutzerdefinierten zustandsbehafteten Anwendung.

Hinweis

RocksDB ist der Standardstatusanbieter ab Databricks Runtime 17.2. Wenn die Abfrage aufgrund einer Ausnahme eines nicht unterstützten Anbieters fehlschlägt, fügen Sie die folgenden Pipelinekonfigurationen hinzu, führen Sie eine vollständige Aktualisierungs- oder Prüfpunktzurücksetzung durch und führen Sie dann die Pipeline erneut aus:

"configuration": {
    "spark.sql.streaming.stateStore.providerClass": "com.databricks.sql.streaming.state.RocksDBStateStoreProvider",
    "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled": "true"
}

from typing import Iterator

import pandas as pd

from pyspark import pipelines as dp
from pyspark.sql.functions import col, from_json
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType, StringType, TimestampType

KAFKA_TOPIC = "<your-kafka-topic>"

output_schema = StructType([
    StructField("sensor_id", LongType(), False),
    StructField("sensor_type", StringType(), False),
    StructField("last_heartbeat_time", TimestampType(), False)])

class SensorHeartbeatProcessor(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        # Define state schema to store sensor information (sensor_id is the grouping key)
        state_schema = StructType([
            StructField("sensor_type", StringType(), False),
            StructField("last_heartbeat_time", TimestampType(), False)])
        self.sensor_state = handle.getValueState("sensorState", state_schema)
        # State variable to track the previously registered timer
        timer_schema = StructType([StructField("timer_ts", LongType(), False)])
        self.timer_state = handle.getValueState("timerState", timer_schema)
        self.handle = handle

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        # Process one row from input and update state
        pdf = next(rows)
        row = pdf.iloc[0]
        # Store or update the sensor information in state using current timestamp
        current_time = pd.Timestamp(timerValues.getCurrentProcessingTimeInMs(), unit='ms')
        self.sensor_state.update((
            row["sensor_type"],
            current_time
        ))

        # Delete old timer if already registered
        if self.timer_state.exists():
            old_timer = self.timer_state.get()[0]
            self.handle.deleteTimer(old_timer)

        # Register a timer for 5 minutes from current processing time
        expiry_time = timerValues.getCurrentProcessingTimeInMs() + (5 * 60 * 1000)
        self.handle.registerTimer(expiry_time)
        # Store the new timer timestamp in state
        self.timer_state.update((expiry_time,))

        # No output on input processing, output only on timer expiry
        return iter([])

    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo) -> Iterator[pd.DataFrame]:
        # Emit output row based on state store
        if self.sensor_state.exists():
            state = self.sensor_state.get()
            output = pd.DataFrame({
                "sensor_id": [key[0]],  # Use grouping key as sensor_id
                "sensor_type": [state[0]],
                "last_heartbeat_time": [state[1]]
            })
            # Remove the entry for the sensor from the state store
            self.sensor_state.clear()
            # Remove the timer state entry
            self.timer_state.clear()
            yield output

    def close(self) -> None:
        pass

dp.create_streaming_table("sensorAlerts")

# Define the schema for the Kafka message value
sensor_schema = StructType([
    StructField("sensor_id", LongType(), False),
    StructField("sensor_type", StringType(), False),
    StructField("sensor_value", LongType(), False)])

@dp.append_flow(target = "sensorAlerts")
def kafka_delta_flow():
    return (
      spark.readStream
        .format("kafka")
        .option("subscribe", KAFKA_TOPIC)
        .option("startingOffsets", "earliest")
        .load()
        .select(from_json(col("value").cast("string"), sensor_schema).alias("data"), col("timestamp"))
        .select("data.*", "timestamp")
        .withWatermark('timestamp', '1 hour')
        .groupBy(col("sensor_id"))
        .transformWithStateInPandas(
          statefulProcessor = SensorHeartbeatProcessor(),
          outputStructType = output_schema,
          outputMode = 'update',
          timeMode = 'ProcessingTime'))

Feedback

War diese Seite hilfreich?

Last updated on 2026-07-10

Flüsse in Lakeflow-Pipelines verwenden

Beispiel: Erstellen eines Standardflusses

SQL

Python

Beispiel: Definieren eines Flusses getrennt von seinem Ziel

Python

SQL

Beispiel: Schreiben in eine Streamingtabelle aus mehreren Kafka-Themen

Python

SQL

Beispiel: Ausführen eines einmaligen Datenrückfüllens

Python

SQL

Beispiel: Verwenden Sie die Anfügeflussverarbeitung anstelle von UNION

Python

SQL

Beispiel: Verwenden Sie transformWithState zum Überwachen von Sensor-Heartbeats

Feedback

Zusätzliche Ressourcen

Beispiel: Verwenden Sie die Anfügeflussverarbeitung anstelle von `UNION`

Beispiel: Verwenden Sie `transformWithState` zum Überwachen von Sensor-Heartbeats