Configurer le mode temps réel

Cette page décrit comment exécuter des query en mode temps réel directement avec les APIs Structured Streaming sur un compute classique. Pour un tutoriel étape par étape, consultez Tutoriel : Exécuter une charge de travail de streaming en temps réel. Pour obtenir des informations conceptuelles sur le mode temps réel, consultez Concepts du mode temps réel.

Le mode temps réel dans les LakeFlow Pipelines prend en charge les pipelines gérés sur les computes classiques et Serverless, avec des exigences distinctes en termes de runtime, de sources, de puits et d’opérateurs.

Prérequis

Pour utiliser le mode temps réel, vous devez configurer votre compute afin de satisfaire aux exigences suivantes :

• Utilisez le compute classique. Les modes d’accès dédié et standard sont pris en charge. Le mode d’accès standard est pris en charge uniquement pour Python. La configuration de la query directe sur cette page n’est pas prise en charge dans les Notebooks ou les Job Serverless. Pour le traitement en temps réel Serverless, utilisez le mode temps réel dans les LakeFlow Pipelines.
• Utiliser Databricks Runtime 16.4 LTS et versions supérieures.
• Désactiver la mise à l'échelle automatique.
• Désactiver Photon.
• Définir spark.databricks.streaming.realTimeMode.enabled sur true.
• Désactivez les instances ponctuelles pour éviter les interruptions.

Pour les charges de travail sensibles à la latence avec des UDF, Databricks vous recommande d'utiliser le mode d'accès dédié. Voir Fonctions de table.

Pour obtenir des instructions sur la création et la configuration du compute classique, consultez Référence de configuration du compute.

Jointures Stream à Stream

Les jointures internes stream-to-stream nécessitent une configuration supplémentaire pour le mode temps réel. Les jointures externes stream-to-stream ne sont pas prises en charge. Consultez l’article dédié à la jointure stream-to-stream.

important

Pour exécuter une jointure Stream to Stream en mode temps réel avec plusieurs autres Stream sur le même cluster, vous devez utiliser Databricks Runtime 18 LTS ou une version ultérieure.

Dans Databricks Runtime 18.2 et versions antérieures, Structured Streaming ne prend pas en charge les configurations suivantes pour les autres modes de traitement, notamment processingTime et availableNow.

Pour activer les jointures Stream vers Stream en mode temps réel, définissez les configurations Spark suivantes :

Python

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

Scala

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

SQL

SQL

SET spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled = true;
SET spark.sql.streaming.join.stateFormatVersion = 4;
SET spark.sql.streaming.join.stateFormatV4.enabled = true;
SET spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion = 2;
SET spark.sql.streaming.realTimeMode.controlMessage.enabled = true;

Configuration de la query

Pour exécuter une query en mode temps réel, vous devez activer le Trigger temps réel. Les Trigger temps réel ne sont pris en charge qu’en mode mise à jour.

Python

Python

query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .outputMode("update")
        # In PySpark, the realTime trigger requires specifying the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Scala

Scala

import org.apache.spark.sql.execution.streaming.RealTimeTrigger

val readStream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic).load()
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .outputMode("update")
      .trigger(RealTimeTrigger.apply())
      // RealTimeTrigger can also accept an argument specifying the checkpoint interval.
      // For example, this code indicates a checkpoint interval of 5 minutes:
      // .trigger(RealTimeTrigger.apply("5 minutes"))
      .start()

Dimensionnement du compute

Vous pouvez exécuter un Job en temps réel par ressource de compute si le compute dispose de suffisamment d'emplacements de tâches.

Pour fonctionner en mode faible latence, le nombre total d'emplacements de tâche disponibles doit être supérieur ou égal au nombre de tâches à travers toutes les étapes de query.

Exemples de calcul d'emplacement

Type de pipeline	Configuration	Emplacements requis
Sans état à un seul étage (source Kafka + sink)	maxPartitions = 8	8 emplacements
À deux étapes et avec état (source Kafka + brassage)	maxPartitions = 8, partitions de shuffle = 20	28 slots (8 + 20)
À trois étapes (source Kafka + shuffle + repartition)	maxPartitions = 8, deux étapes de brassage de 20 chacune	48 emplacements (8 + 20 + 20)

Type de pipeline	Configuration	Emplacements requis
Sans état à un seul étage (source Kafka + sink)	maxPartitions = 8	8 emplacements
À deux étapes et avec état (source Kafka + brassage)	maxPartitions = 8, partitions de shuffle = 20	28 slots (8 + 20)
À trois étapes (source Kafka + shuffle + repartition)	maxPartitions = 8, deux étapes de brassage de 20 chacune	48 emplacements (8 + 20 + 20)

Si vous ne définissez pas maxPartitions, utilisez le nombre de partitions dans le sujet Kafka.

Ressources supplémentaires

• Tutoriel : exécuter une charge de travail de streaming en temps réel
• Exemples de mode temps réel
• Optimiser et surveiller les performances des query en mode temps réel
• Référence du mode Temps réel