Aller au contenu principal
Dernière mise à jour le

Exemples en mode temps réel

Cette page fournit des exemples de code fonctionnels pour les queries en mode temps réel dans le Structured Streaming, des transformations sans état simples au traitement avec état complexe avec gestion d'état personnalisée. Consultez le mode temps réel dans le Structured Streaming pour les concepts et le tutoriel : Exécuter une charge de travail de streaming en temps réel pour un tutoriel pratique.

Prérequis​

Pour exécuter les exemples sur cette page, vous avez besoin :

  • Un cluster du mode temps réel configuré et en cours d'exécution. Consultez Configurer le mode temps réel pour connaître les exigences et la configuration, ou Tutoriel : Exécuter une charge de travail de streaming en temps réel pour un tutoriel pas à pas.
  • Familiarité de base avec les concepts de Structured Streaming. Consultez les concepts de Structured Streaming si vous débutez avec le streaming.
  • Accès aux sources et récepteurs de streaming pris en charge :
    • Pour les exemples Kafka : Un broker Kafka avec des sujets d'entrée/sortie configurés
    • Pour les exemples Kinesis : informations d'identification AWS et un Stream Kinesis configuré pour le mode de diffusion améliorée (EFO)
    • Pour les exemples Lakebase : une base de données Lakebase. Consultez les Exigences pour connaître les exigences.
    • Pour les exemples de récepteurs personnalisés : Base de données ou service cible configuré (PostgreSQL pour l'exemple fourni)
remarque

Les exemples utilisent des valeurs d'espace réservé comme broker_address, input_topic et checkpoint_location. Remplacez-les par vos valeurs de configuration réelles avant d'exécuter le code.

Exemples de query sans état​

Les query sans état traitent chaque enregistrement indépendamment sans maintenir d'état entre les enregistrements. Ces query sont généralement plus simples et ont une latence plus faible que les query avec état, car elles n'ont pas besoin de gérer le stockage d'état ou d'effectuer des recherches. Utilisez des query sans état pour les transformations, le filtrage, les jointures avec des données statiques et les opérations de routage.

Du Kafka source au Kafka sink​

Dans cet exemple, vous lisez à partir d'une source Kafka et écrivez vers un récepteur Kafka.

Python
query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("startingOffsets", "earliest")
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .trigger(realTime="5 minutes")
        .outputMode("update")
        .start()
)

Repartitionnement​

Dans cet exemple, vous lisez à partir d'une source Kafka, repartitionnez les données en 20 partitions et écrivez dans un récepteur Kafka.

En raison d'une limitation d'implémentation actuelle, vous devez définir la configuration Spark spark.sql.execution.sortBeforeRepartition sur false avant d'utiliser la répartition.

Python
# Sorting is not supported in repartition with real-time mode, so you must set this to false to achieve low latency.
spark.conf.set("spark.sql.execution.sortBeforeRepartition", "false")

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("subscribe", input_topic)
    .option("startingOffsets", "earliest")
    .load()
    .repartition(20)
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Jointure Stream-snapshot (diffusion uniquement)​

Dans cet exemple, vous lisez des données depuis Kafka, joignez les données à une table statique et écrivez dans un récepteur Kafka. Seules les jointures stream-statiques qui diffusent la table statique sont prises en charge, ce qui signifie que la table statique doit tenir en mémoire.

Python
from pyspark.sql.functions import broadcast, expr

# We assume the static table in the path `static_table_location` has a column 'lookupKey'.

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("subscribe", input_topic)
    .option("startingOffsets", "earliest")
    .load()
    .withColumn("joinKey", expr("CAST(value AS STRING)"))
    .join(
        broadcast(spark.read.format("parquet").load(static_table_location)),
        expr("joinKey = lookupKey")
    )
    .selectExpr("value AS key", "value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Source Kinesis vers récepteur Kafka​

Dans cet exemple, vous lisez à partir d'une source Kinesis et écrivez dans un récepteur Kafka.

Python
query = (
    spark.readStream
        .format("kinesis")
        .option("region", region_name)
        .option("awsAccessKey", aws_access_key_id)
        .option("awsSecretKey", aws_secret_access_key)
        .option("consumerMode", "efo")
        .option("consumerName", consumer_name)
        .load()
        .selectExpr("partitionKey AS key", "CAST(data AS STRING) AS value")
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .trigger(realTime="5 minutes")
        .outputMode("update")
        .start()
)

Union​

Dans cet exemple, vous unissez deux DataFrames Kafka de deux sujets différents et écrivez dans un sink Kafka.

Python
df1 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_1)
    .load()
)

df2 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_2)
    .load()
)

query = (
    df1.union(df2)
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Source Kafka vers récepteur Lakebase​

Dans ces exemples, vous lisez à partir d'une source Kafka et écrivez dans une base de données Lakebase. Pour toutes les options de connexion, consultez Connectez-vous à Lakebase.

Tables Lakebase enregistrées avec Unity Catalog​

L'exemple suivant utilise la méthode de connexion Unity Catalog, qui gère automatiquement les identifiants. La base de données Lakebase cible doit être enregistrée auprès de Unity Catalog. Voir Enregistrer une base de données Lakebase dans Unity Catalog.

Python
query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .option("startingOffsets", "earliest")
        .load()
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .writeStream
        .outputMode("update")
        .option("checkpointLocation", checkpoint_location)
        .option("upsertkey", "<primary_key>")
        .trigger(realTime="5 minutes")
        .toTable("<catalog>.<schema>.<table>")
)

Remplacez <catalog>.<schema>.<table> par le nom entièrement qualifié de la table cible et <primary_key> par la colonne de clé primaire. Si la table n'existe pas, le connecteur la crée à l'aide de cette clé.

Tables Lakebase non enregistrées avec Unity Catalog​

Pour les tables Lakebase non enregistrées avec Unity Catalog, utilisez les options endpoint et dbtable :

Python
query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .option("startingOffsets", "earliest")
        .load()
        .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .writeStream
        .format("postgresql")
        .outputMode("update")
        .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
        .option("dbtable", "<schema>.<table>")
        .option("upsertkey", "<primary_key>")
        .option("checkpointLocation", checkpoint_location)
        .trigger(realTime="5 minutes")
        .start()
)

Remplacez <catalog>.<schema>.<table> par le nom entièrement qualifié de la table cible, et <primary_key> par la colonne de clé primaire.

Consultez Tables Lakebase non enregistrées auprès d'Unity Catalog pour toutes les options disponibles.

Exemples de requêtes avec état​

Les queries avec état conservent les informations d'état entre les enregistrements, permettant des Opérations telles que la déduplication, l'agrégation et le fenêtrage. Ces queries sont essentielles pour les cas d'utilisation qui nécessitent le suivi des informations dans le temps ou à travers plusieurs événements. Le mode temps réel prend en charge les Opérations avec état avec la même sémantique que le mode micro-batch, mais traite les données en continu pour une latence plus faible. Les requêtes avec état nécessitent davantage de mémoire et de Ressources de compute que les requêtes sans état, car elles doivent maintenir et mettre à jour leur état.

Déduplication​

Dans cet exemple, vous dédupliquez les enregistrements en fonction des colonnes timestamp et value.

Python
query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic)
    .load()
    .dropDuplicates(["timestamp", "value"])
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Agrégation​

Dans cet exemple, vous regroupez les enregistrements par timestamp et value, puis comptez les occurrences.

Python
from pyspark.sql.functions import col

query = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic)
    .load()
    .groupBy(col("timestamp"), col("value"))
    .count()
    .selectExpr("CAST(value AS STRING) AS key", "CAST(count AS STRING) AS value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

Union avec agrégation​

Dans cet exemple, vous unissez d'abord deux DataFrames Kafka provenant de deux sujets différents, puis vous effectuez une agrégation. Au final, vous écrivez dans le récepteur Kafka.

Python
from pyspark.sql.functions import col

df1 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_1)
    .load()
)

df2 = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("startingOffsets", "earliest")
    .option("subscribe", input_topic_2)
    .load()
)

query = (
    df1.union(df2)
    .groupBy(col("timestamp"), col("value"))
    .count()
    .selectExpr("CAST(value AS STRING) AS key", "CAST(count AS STRING) AS value")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", broker_address)
    .option("topic", output_topic)
    .option("checkpointLocation", checkpoint_location)
    .trigger(realTime="5 minutes")
    .outputMode("update")
    .start()
)

transformWithState​

Dans cet exemple, vous utilisez transformWithState pour maintenir un état personnalisé avec un TTL (durée de vie). Le processeur compte le nombre d’enregistrements vus pour chaque clé.

Python
from typing import Iterator, Tuple

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import LongType, StringType, TimestampType, StructField, StructType


class RTMStatefulProcessor(StatefulProcessor):
  """
  This processor counts the number of records it has seen for each key using state variables
  with TTLs. It redundantly maintains this count with a value, list, and map state to put load
  on the state variable cleanup mechanism. (In practice, only one value state is needed to maintain
  the count for a given grouping key.)

  The input schema it expects is (String, Long) which represents a (key, source-timestamp) tuple.
  The source-timestamp is passed through so that we can calculate end-to-end latency. The output
  schema is (String, Long, Long), which represents a (key, count, source-timestamp) 3-tuple.
  """

  def init(self, handle: StatefulProcessorHandle) -> None:
    state_schema = StructType([StructField("value", LongType(), True)])
    self.value_state = handle.getValueState("value", state_schema, 30000)
    map_key_schema = StructType([StructField("key", LongType(), True)])
    map_value_schema = StructType([StructField("value", StringType(), True)])
    self.map_state = handle.getMapState("map", map_key_schema, map_value_schema, 30000)
    list_schema = StructType([StructField("value", StringType(), True)])
    self.list_state = handle.getListState("list", list_schema, 30000)

  def handleInputRows(self, key, rows, timerValues) -> Iterator[Row]:
    for row in rows:
      # row is a tuple (key, source_timestamp)
      key_str = row[0]
      source_timestamp = row[1]
      old_value = value.get()
      if old_value is None:
        old_value = 0
      self.value_state.update((old_value + 1,))
      self.map_state.update((old_value,), (key_str,))
      self.list_state.appendValue((key_str,))
      yield Row(key=key_str, value=old_value + 1, timestamp=source_timestamp)

  def close(self) -> None:
    pass


output_schema = StructType(
  [
    StructField("key", StringType(), True),
    StructField("value", LongType(), True),
    StructField("timestamp", TimestampType(), True),
  ]
)

query = (
  spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", broker_address)
  .option("subscribe", input_topic)
  .load()
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
  .groupBy("key")
  .transformWithState(
    statefulProcessor=RTMStatefulProcessor(),
    outputStructType=output_schema,
    outputMode="Update",
    timeMode="processingTime",
  )
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", broker_address)
  .option("topic", output_topic)
  .option("checkpointLocation", checkpoint_location)
  .trigger(realTime="5 minutes")
  .outputMode("Update")
  .start()
)
remarque

Il existe une différence dans la manière dont le mode temps réel et les autres modes d’exécution de Structured Streaming exécutent le StatefulProcessor dans le transformWithState. Veuillez consulter transformWithState en mode temps réel.

Développement et tests​

Vous pouvez utiliser la fonction display pour visualiser les données de streaming en temps réel directement dans un Notebook et pour vérifier votre logique de query et vos Transformations de données avant de déployer en production avec Kafka ou des récepteurs personnalisés. Ceci est utile pour le développement interactif, les tests et le debugging des queries en mode temps réel sans configurer de récepteurs externes ou d'infrastructure de production.

La fonction display avec le trigger realTime est disponible dans Databricks Runtime 17.1 et versions ultérieures. Pour un exemple complet utilisant la source de débit avec display, consultez Tutoriel : Exécuter une charge de travail de streaming en temps réel.

Afficher la source du taux​

Dans cet exemple, vous lisez à partir d'une source de débit et affichez le DataFrame de streaming dans un notebook.

Python
inputDF = (
  spark
  .readStream
  .format("rate")
  .option("numPartitions", 2)
  .option("rowsPerSecond", 1)
  .load()
)
display(inputDF, realTime="5 minutes", outputMode="update")

Exemples de sink personnalisés​

Lorsque vous devez écrire des données de streaming vers des destinations qui ne prennent pas en charge le Structured Streaming intégré, utilisez foreachSink pour implémenter une logique d'écriture personnalisée. Les récepteurs personnalisés vous donnent un contrôle total sur la manière dont les données sont écrites, vous permettant de vous intégrer à n'importe quelle base de données, API ou système de stockage. L'exemple suivant montre l'écriture dans une base de données PostgreSQL à l'aide de JDBC.

Écrire dans PostgreSQL en utilisant foreachSink​

Scala
import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.sql.{ForeachWriter, Row}

/**
 * Groups connection properties for
 * the JDBC writers.
 *
 * @param url JDBC url of the form jdbc:subprotocol:subname to connect to
 * @param dbtable database table that should be written into
 * @param username username for authentication
 * @param password password for authentication
 */
class JdbcWriterConfig(
    val url: String,
    val dbtable: String,
    val username: String,
    val password: String,
) extends Serializable

/**
 * Handles streaming data writes to a database sink via JDBC, by:
 *  - connecting to the database
 *  - buffering incoming data rows in batches to reduce write overhead
 *
 * @param config connection parameters and configuration knobs for the writer
 */
class JdbcStreamingDataWriter(config: JdbcWriterConfig)
  extends ForeachWriter[Row] with Serializable {
  // The writer currently only supports this hard-coded schema
  private val UPSERT_STATEMENT_SQL =
    s"""MERGE INTO "${config.dbtable}"
       |USING (
       |  SELECT
       |    CAST(? AS INTEGER) AS "id",
       |    CAST(? AS CHARACTER VARYING) AS "data"
       |) AS "source"
       |ON "test"."id" = "source"."id"
       |WHEN MATCHED THEN
       |  UPDATE SET "data" = "source"."data"
       |WHEN NOT MATCHED THEN
       |  INSERT ("id", "data") VALUES ("source"."id", "source"."data")
       |""".stripMargin

  private val MAX_BUFFER_SIZE = 3
  private val buffer = new Array[Row](MAX_BUFFER_SIZE)
  private var bufferSize = 0

  private var connection: Connection = _

  /**
   * Flushes the [[buffer]] by writing all rows in the buffer to the database.
   */
  private def flushBuffer(): Unit = {
    require(connection != null)

    if (bufferSize == 0) {
      return
    }

    var upsertStatement: PreparedStatement = null

    try {
      upsertStatement = connection.prepareStatement(UPSERT_STATEMENT_SQL)

      for (i <- 0 until bufferSize) {
        val row = buffer(i)
        upsertStatement.setInt(1, row.getAs[String]("key"))
        upsertStatement.setString(2, row.getAs[String]("value"))
        upsertStatement.addBatch()
      }

      upsertStatement.executeBatch()
      connection.commit()

      bufferSize = 0
    } catch { case e: Exception =>
      if (connection != null) {
        connection.rollback()
      }
      throw e
    } finally {
      if (upsertStatement != null) {
        upsertStatement.close()
      }
    }
  }

  override def open(partitionId: Long, epochId: Long): Boolean = {
    connection = DriverManager.getConnection(config.url, config.username, config.password)
    true
  }

  override def process(row: Row): Unit = {
    buffer(bufferSize) = row
    bufferSize += 1
    if (bufferSize >= MAX_BUFFER_SIZE) {
      flushBuffer()
    }
  }

  override def close(errorOrNull: Throwable): Unit = {
    flushBuffer()
    if (connection != null) {
      connection.close()
      connection = null
    }
  }
}


spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", testUtils.brokerAddress)
      .option("subscribe", inputTopic)
      .load()
      .writeStream
      .outputMode(OutputMode.Update())
      .trigger(defaultTrigger)
      .foreach(new JdbcStreamingDataWriter(new JdbcWriterConfig(jdbcUrl, tableName, jdbcUsername, jdbcPassword)))
      .start()

Ressources supplémentaires​

Maintenant que vous avez exploré ces exemples en mode temps réel, voici des ressources pour approfondir vos connaissances et créer des applications de streaming prêtes pour la production :

Sur cette page