Utiliser le mode en temps réel dans les pipelines Lakeflow

Le mode en temps réel permet le traitement des données ultra-à-faible latence, avec une latence de bout en bout aussi faible que cinq millisecondes. Utilisez le mode en temps réel pour les charges de travail opérationnelles qui nécessitent une réponse immédiate aux données de diffusion en continu, telles que la détection des fraudes et la personnalisation en temps réel.

Le mode en temps réel est également disponible directement dans Structured Streaming en dehors des pipelines. Voir le mode temps réel dans Structured Streaming.

Comment le mode en temps réel atteint une faible latence

Le mode en temps réel diffère du traitement continu standard de trois façons clés :

• Lots de longue durée : le système traite les données au fur et à mesure qu’il devient disponible dans la source dans les lots de longue durée (la valeur par défaut est de cinq minutes).
• Planification simultanée des étapes : toutes les phases de requête sont planifiées en même temps. La ressource de calcul doit avoir suffisamment d’emplacements de tâches disponibles pour couvrir toutes les étapes simultanément. Voir Dimensionnement du calcul.
• Shuffle continu : les données sont transmises entre les étapes dès qu’elles sont produites, au lieu d’attendre qu’une étape en amont soit terminée avant de démarrer l’étape en aval.

L’intervalle de point de contrôle (configuré via pipelines.trigger.interval) contrôle la fréquence à laquelle l'état et les offsets de source sont conservés dans un stockage durable. Les intervalles plus longs réduisent la surcharge de point de contrôle, mais augmentent le temps de récupération après un échec et retardent les rapports de métriques. Les intervalles plus courts améliorent la durabilité, mais ajoutent une surcharge.

Mode temps réel et pipelines continus

Le mode en temps réel est un type spécialisé de déclencheur continu. Le mode continu est toujours requis ; Le mode en temps réel ajoute des optimisations de latence au niveau du flux en plus. Pour utiliser le mode en temps réel, le pipeline doit d’abord s’exécuter en mode continu. Le mode en temps réel applique ensuite des optimisations supplémentaires au niveau du flux pour atteindre une latence inférieure à seconde au-delà de ce que fournit le traitement continu standard.

L’activation du mode en temps réel nécessite trois étapes de configuration :

1. Configurez le pipeline en mode continu.
2. Activez le mode en temps réel au niveau du pipeline.
3. Définissez un flux de mise à jour en temps réel.

Requirements

Configurer le mode en temps réel

Étape 1 : Définir le pipeline en mode continu

Dans vos paramètres de pipeline, définissez le mode pipeline sur Continu ou définissez-le dans le json du pipeline :

{
  "continuous": true
}

Étape 2 : Activer le mode en temps réel au niveau du pipeline

Dans vos paramètres de pipeline, ajoutez la clé suivante à la configuration Spark sous configuration Spark avancée >:

spark.databricks.streaming.realTimeMode.enabled = true

Vous pouvez également définir ceci dans le json du pipeline :

{
  "continuous": true,
  "spark_conf": {
    "spark.databricks.streaming.realTimeMode.enabled": "true"
  }
}

Étape 3 : Définir un flux de mise à jour en temps réel

Le mode en temps réel nécessite un flux de mise à jour. Utilisez dp.create_sink() ou @dp.foreach_sink pour définir la cible de sortie, puis utilisez le décorateur @dp.update_flow avec pipelines.trigger défini sur "RealTime" et target pointant vers le collecteur.

from pyspark import pipelines as dp

# Define the output sink
dp.create_sink(
    "my_kafka_sink",
    "kafka",
    {
        "kafka.bootstrap.servers": "<bootstrap-servers>",
        "topic": "<output-topic>",
    }
)

# Define the real-time update flow targeting the sink
@dp.update_flow(
    name="my_rtm_flow",
    target="my_kafka_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",  # optional; defaults to 5 minutes
    }
)
def my_real_time_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "<bootstrap-servers>")
            .option("subscribe", "<input-topic>")
            .load()
    )

Paramètres de configuration au niveau du flux :

Exemples de code

Kafka vers Kafka

Lisez à partir d’un topic Kafka et écrivez vers une cible de sortie Kafka :

from pyspark import pipelines as dp

dp.create_sink("kafka_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="kafka_rtm_flow",
    target="kafka_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def kafka_rtm_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .option("startingOffsets", "latest")
            .load()
            .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
    )

Kafka dans un récepteur ForEach

Lisez un topic Kafka et gérez chaque ligne avec une fonction Python dans un ForEach sink :

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_foreach_sink")
def handle_event(row):
    # Custom per-row logic here, such as a call to an external API.
    ...

@dp.update_flow(
    name="foreach_rtm_flow",
    target="events_foreach_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def foreach_rtm_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .option("startingOffsets", "latest")
            .load()
            .selectExpr("CAST(key AS STRING) AS key", "CAST(value AS STRING) AS value")
    )

Enrichir avec une jointure de diffusion

Joignez un flux Kafka à une table de recherche statique. Seules les jointures de diffusion (flux à statique) sont prises en charge. Les jointures de flux à flux ne sont pas prises en charge en mode temps réel.

from pyspark import pipelines as dp
from pyspark.sql.functions import broadcast, expr

dp.create_sink("enriched_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": enriched_output_topic,
})

@dp.update_flow(
    name="enriched_events_flow",
    target="enriched_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def enriched_events():
    lookup = spark.read.table("catalog.schema.lookup_table")
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .withColumn("event_key", expr("CAST(value AS STRING)"))
            .join(broadcast(lookup), expr("event_key = lookup_key"))
            .select("event_key", "lookup_value", "timestamp")
    )

Agrégation

Compter les événements par clé avec état groupBy. Définissez spark.sql.shuffle.partitions de manière à correspondre au nombre de partitions d’entrée pour les opérations avec état :

from pyspark import pipelines as dp
from pyspark.sql.functions import col

dp.create_sink("event_counts_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="event_counts_flow",
    target="event_counts_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
        "spark.sql.shuffle.partitions": "8",
    }
)
def event_counts():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .selectExpr("CAST(key AS STRING) AS event_type", "timestamp")
            .groupBy(col("event_type"))
            .count()
    )

Sources et puits pris en charge

Dimensionnement du calcul

Pipelines sans serveur sur le canal d’aperçu

Azure Databricks réserve des emplacements de tâches pour chaque flux en temps réel, ce qui permet d’exécuter plusieurs flux en temps réel dans un même pipeline sans que les flux ne se disputent les emplacements de tâches. Tu n’as pas besoin de dimensionner les ressources de calcul.

Le repartitionnement adaptatif de l’étape source ajuste automatiquement le nombre de tâches qui lisent de chaque source Kafka ou Kinesis pour correspondre à son trafic. Pour l’utiliser, laissez maxPartitions non défini pour les sources Kafka, et laissez maxPartitions et shardsPerTask non définis pour les sources Kinesis. Si vous définissez ces options, la source utilise le nombre fixe de tâches que vous avez configurées à la place. Le nombre de partitions de shuffle n’est pas ajusté, donc définissez spark.sql.shuffle.partitions pour des flux avec état, comme indiqué dans l’agrégation.

Pipelines sans serveur sur le canal actuel

Azure Databricks dimensionne les ressources de calcul pour vous. Le nombre de tâches qui lisent depuis chaque source est fixe, donc définissez maxPartitions sur des sources Kafka, ou maxPartitions ou shardsPerTask sur des sources Kinesis pour gérer votre trafic maximal.

Calcul classique

Le compute doit disposer de suffisamment de slots de tâches pour exécuter toutes les tâches dans toutes les étapes de la requête simultanément. Le tableau suivant montre les emplacements de tâches requis pour les types courants de pipelines :

Si vous ne définissez maxPartitionspas, utilisez le nombre de partitions dans la rubrique Kafka.

Prise en charge des opérateurs

transformWithState en mode temps réel

transformWithState est pris en charge en mode en temps réel avec les différences suivantes par rapport au traitement par micro-lots :

• handleInputRows est appelé une fois par ligne plutôt qu’une fois par clé par lot. L’itérateur inputRows génère une valeur unique par appel.
• Les minuteries de temps d'événements ne sont pas prises en charge. Les minuteurs en temps de traitement se déclenchent lorsqu’un traitement par lot de longue durée se termine, si aucune donnée n’est arrivée.
• La fonction transformWithStateInPandas n'est pas prise en charge.

UDF Pandas en mode en temps réel

Pour minimiser la latence avec les UDF pandas, définissez spark.sql.execution.arrow.maxRecordsPerBatch sur 1. Cela optimise la latence au détriment du débit. Si le débit est également important, définissez cette valeur sur 100 ou une valeur supérieure.

Surveiller les performances du mode en temps réel

Le mode en temps réel affiche les métriques de latence dans le champ StreamingQueryProgress de latencies. Accédez à ces métriques via une StreamingQueryListener ou en inspectant la lastProgress propriété sur la requête de diffusion en continu.

Chaque métrique est signalée comme p50, p90, p95 et p99 centiles.

Limitations

Sur le calcul classique et les pipelines serverless utilisant le canal actuel, un flux temps réel par pipeline est recommandé. Plusieurs flux sont autorisés, mais ils se disputent les emplacements de tâches, ce qui augmente la latence. Pour exécuter plusieurs flux en temps réel dans un même pipeline sans cette augmentation de latence, voir Pipelines serverless sur le canal de prévisualisation.

Pour obtenir la liste complète des limitations de l’opérateur et de la source, consultez les limitations du mode en temps réel.

Ressources additionnelles

• Concepts de mode temps réel
• Référence du mode en temps réel
• Mode de pipeline déclenché ou continu
• Configurer le calcul classique pour les pipelines