Utiliser des flux dans les pipelines Lakeflow

Les flux d’un pipeline Lakeflow acheminent les données vers une table de flux ou une vue matérialisée. Les exemples suivants montrent comment définir des flux par défaut, définir un flux séparément de sa cible, écrire dans une table de diffusion en continu à partir de plusieurs rubriques Kafka, exécuter un remplissage unique et remplacer UNION des requêtes par un traitement de flux d’ajout.

Pour obtenir une vue d’ensemble des flux, consultez Charger et traiter des données de manière incrémentielle avec les flux de pipeline Lakeflow.

Exemple : Créer un flux par défaut

Lorsque vous créez un pipeline, vous définissez généralement une table ou une vue avec la requête qui la prend en charge. Par exemple, cette requête crée une table de streaming nommée customers_silver à partir de customers_bronze. La table de diffusion en continu et son flux par défaut sont créés ensemble en une seule étape.

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

Le flux par défaut d’une table de diffusion en continu est un flux d’ajout qui ajoute de nouvelles lignes avec chaque mise à jour et a le même nom que la cible. Il s’agit de la façon la plus courante d’utiliser des pipelines ( création d’un flux et de sa cible en une seule étape) et vous pouvez l’utiliser pour ingérer ou transformer des données. Pour plus d’informations sur les concepts de flux, consultez Charger et traiter des données de manière incrémentielle avec les flux de pipeline Lakeflow.

Exemple : Définir un flux séparément de sa cible

Vous pouvez également créer un flux pour une table que vous avez définie séparément. Le résultat est identique à la création d’un flux par défaut, notamment en utilisant le même nom pour la table de diffusion en continu et le flux :

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

La définition d’un flux séparément de sa cible vous permet de créer plusieurs flux qui ajoutent des données à la même cible. Utilisez le @dp.append_flow décorateur dans l’interface Python ou la CREATE FLOW...INSERT INTO clause de l’interface SQL pour ajouter des flux pour des tâches telles que les suivantes :

Pour les requêtes Python, utilisez la fonction create_streaming_table() pour créer une table cible.

Important

  • Si vous devez définir des contraintes de qualité des données avec des attentes, définissez les attentes sur la table cible dans le cadre de la create_streaming_table() fonction ou sur une définition de table existante. Vous ne pouvez pas définir les attentes dans la @append_flow définition.
  • Les flux sont identifiés par un nom de flux et ce nom est utilisé pour identifier les points de contrôle de diffusion en continu. L’utilisation du nom de flux pour identifier le point de contrôle signifie ce qui suit :
    • Si un flux existant dans un pipeline est renommé, le point de contrôle n’est pas reporté et le flux renommé est effectivement un flux entièrement nouveau.
    • Vous ne pouvez pas réutiliser un nom de flux dans un pipeline, car le point de contrôle existant ne correspond pas à la nouvelle définition de flux.

Voir l’exemple : Écrire dans une table de diffusion en continu à partir de plusieurs rubriques Kafka.

Les exemples suivants créent une table de diffusion en continu nommée kafka_target et écrit dans cette table de diffusion en continu à partir de deux rubriques Kafka :

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');

Pour en savoir plus sur la read_kafka() fonction à valeur de table utilisée dans les requêtes SQL, consultez read_kafka dans la documentation SQL.

En Python, vous pouvez créer par programmation plusieurs flux qui ciblent une seule table. L’exemple suivant montre ce modèle pour une liste de rubriques Kafka.

Note

Ce modèle a les mêmes exigences que l’utilisation d’une for boucle pour créer des tables. Vous devez transmettre explicitement une valeur Python à la fonction définissant le flux. Consultez Créer des tables dans une for boucle.

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

Exemple : Exécuter un remplissage de données unique

Si vous souhaitez exécuter une requête pour ajouter des données à une table de diffusion en continu existante, utilisez append_flow.

Après avoir ajouté un ensemble de données existantes, vous avez plusieurs options :

  • Si vous souhaitez que la requête ajoute de nouvelles données lorsqu'elles arrivent dans le répertoire de rétro-remplissage, laissez la requête telle quelle.
  • Si vous souhaitez qu'il s'agisse d'un renvoi unique à ne plus jamais exécuter, supprimez la requête après avoir exécuté le pipeline une seule fois.
  • Si vous souhaitez que la requête s'exécute une seule fois, et ne la réexécuter que dans les cas où les données sont entièrement actualisées, paramétrez le paramètre once à True dans le flux d'ajout. Dans SQL, utilisez INSERT INTO ONCE.

Les exemples suivants exécutent une requête pour ajouter des données historiques à une table de diffusion en continu :

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"
  );

Pour obtenir un exemple plus approfondi, consultez Remplissage des données historiques avec des pipelines.

Exemple : Utiliser le traitement de flux d’ajout au lieu de UNION

Au lieu d’utiliser une requête avec une clause UNION, vous pouvez utiliser des requêtes de flux d’ajout pour combiner plusieurs sources et écrire dans une seule table de diffusion en continu. L’utilisation de requêtes de flux d’ajout au lieu de UNION vous permet d'ajouter à une table de diffusion en continu de plusieurs sources sans exécuter une actualisation complète.

L’exemple Python suivant inclut une requête qui combine plusieurs sources de données avec une UNION clause :

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

Les exemples suivants remplacent la UNION requête par des requêtes de flux d’ajout :

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"
  );

Exemple : Utiliser transformWithState pour surveiller les pulsations de capteur

L’exemple suivant montre un processeur avec état qui lit à partir de Kafka et vérifie que les capteurs émettent régulièrement des pulsations. Si une pulsation n’est pas reçue dans les 5 minutes, le processeur émet une entrée dans la table Delta cible à des fins d’analyse.

Pour plus d’informations sur la création d’applications avec état personnalisées, consultez Générer une application avec état personnalisé.

Note

RocksDB est le fournisseur d’état par défaut à partir de Databricks Runtime 17.2. Si la requête échoue en raison d’une exception de fournisseur non prise en charge, ajoutez les configurations de pipeline suivantes, effectuez une réinitialisation complète ou une réinitialisation du point de contrôle, puis réexécutez votre pipeline :

"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'))
  • Last updated on