trigger (DataStreamWriter)

Définit le Trigger pour la query de streaming. Si non défini, la query s'exécute aussi vite que possible, équivalent à processingTime='0 seconds'. Un seul paramètre de trigger peut être défini à la fois.

Consultez l’article Configurer les intervalles de trigger de Structured Streaming.

Exécution de compute et de pipeline

Dans les LakeFlow Pipelines, le pipeline gère l’exécution du Stream. Utilisez les paramètres d’exécution du pipeline pour contrôler le moment où les données du pipeline sont traitées.

Lorsque vous utilisez DataStreamWriter directement sur un compute Serverless pour les Notebooks et les Job, utilisez availableNow=True. Le trigger default, processingTime, continuous et realTime ne sont pas pris en charge sur ce compute. Ces restrictions ne s’appliquent pas aux modes d’exécution de pipelines serverless. Voir Limitations du streaming.

Syntaxe

trigger(*, processingTime=None, once=None, continuous=None, availableNow=None, realTime=None)

parameter

parameter	Type	Description
processingTime	str, facultatif	Une chaîne d'intervalle de temps de traitement (par exemple, '5 seconds', '1 minute'). Exécute une query de micro-lots périodiquement en fonction du temps de traitement.
once	bool, facultatif	Si True, traite uniquement un batch de données puis met fin à la query. Obsolète à partir de Databricks Runtime 11.3 LTS. Utilisez plutôt availableNow=True.
continuous	str, facultatif	Non pris en charge sur Databricks. Ce parameter active le mode de traitement continu expérimental d’Apache Spark, qui diffère du mode de pipeline continu. Utilisez le mode en temps réel pour les query Structured Streaming à faible latence.
availableNow	bool, facultatif	Si True, traite toutes les données disponibles par lots, puis met fin à la query.
realTime	str, facultatif	Une chaîne de caractères indiquant la durée du batch (par exemple, '5 minutes'). Traite les enregistrements à leur arrivée au sein d'un traitement par batch de longue durée. La valeur contrôle la durée du batch, pas la latence de traitement des enregistrements. Consultez l’article dédié au mode en temps réel de Structured Streaming.

parameter	Type	Description
processingTime	str, facultatif	Une chaîne d'intervalle de temps de traitement (par exemple, '5 seconds', '1 minute'). Exécute une query de micro-lots périodiquement en fonction du temps de traitement.
once	bool, facultatif	Si True, traite uniquement un batch de données puis met fin à la query. Obsolète à partir de Databricks Runtime 11.3 LTS. Utilisez plutôt availableNow=True.
continuous	str, facultatif	Non pris en charge sur Databricks. Ce parameter active le mode de traitement continu expérimental d’Apache Spark, qui diffère du mode de pipeline continu. Utilisez le mode en temps réel pour les query Structured Streaming à faible latence.
availableNow	bool, facultatif	Si True, traite toutes les données disponibles par lots, puis met fin à la query.
realTime	str, facultatif	Une chaîne de caractères indiquant la durée du batch (par exemple, '5 minutes'). Traite les enregistrements à leur arrivée au sein d'un traitement par batch de longue durée. La valeur contrôle la durée du batch, pas la latence de traitement des enregistrements. Consultez l’article dédié au mode en temps réel de Structured Streaming.

Renvoie

DataStreamWriter

Exemples

Python

df = spark.readStream.format("rate").load()

Exécution du Trigger toutes les 5 secondes :

Python

df.writeStream.trigger(processingTime='5 seconds')
# <...streaming.readwriter.DataStreamWriter object ...>

Traitez toutes les données disponibles par lots :

Python

df.writeStream.trigger(availableNow=True)
# <...streaming.readwriter.DataStreamWriter object ...>

Configurez le mode temps réel avec des batchs de cinq minutes. Les enregistrements sont traités à mesure qu’ils arrivent pendant chaque batch. Utilisez un type de source, de destination et de compute pris en charge:

Python

kafka_df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<bootstrap-servers>")
    .option("subscribe", "<input-topic>")
    .load())
kafka_df.writeStream.trigger(realTime="5 minutes")
# <...streaming.readwriter.DataStreamWriter object ...>