トリガー（DataStreamWriter）

ストリーミングクエリのトリガーを設定します。設定されていない場合、クエリは可能な限り高速に実行され、 processingTime='0 seconds'と同等になります。一度に設定できるトリガーは 1 つだけです。

Structured Streaming のTrigger間隔の設定を参照してください。

コンピュートとパイプラインの実行

LakeFlow Pipelinesでは、パイプラインはストリームの実行を管理します。パイプラインデータの処理タイミングを制御するには、パイプライン実行設定を使用してください。

ノートブックとジョブのServerlessコンピュートで DataStreamWriter を直接使用する場合は、 availableNow=True を使用してください。そのコンピュートでは、defaultのトリガー processingTime、 continuous、 realTimeはサポートされていません。これらの制限は、Serverless パイプラインの実行モードには適用されません。ストリーミングの制限事項を参照してください。

構文

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

パラメーター

パラメーター	Type	説明
processingTime	文字列、オプション	処理時間間隔文字列（例： '5 seconds' 、 '1 minute' ）。処理時間に基づいて、マイクロバッチクエリを定期的に実行します。
once	ブール値、オプション	True の場合、データのバッチを 1 つだけ処理し、クエリーを終了します。Databricks Runtime 11.3 LTS以降では非推奨。代わりに availableNow=True を使用してください。
continuous	文字列、オプション	Databricksではサポートされていません。このパラメーターは、連続パイプラインモードとは異なる Apache Spark の実験的な連続処理モードを有効にします。低遅延のStructured Streaming クエリーにはリアルタイムモードを使用します。
availableNow	ブール値、オプション	Trueの場合、利用可能なすべてのデータを複数のバッチで処理してからクエリを終了します。
realTime	文字列、オプション	バッチの期間を表す文字列。例：'5 minutes'。長時間実行されるバッチ内でレコードが到着したときに処理します。この値はバッチの期間を制御するものであり、レコード処理の待機時間を制御するものではありません。Structured Streaming のリアルタイムモードを参照してください。

パラメーター	Type	説明
processingTime	文字列、オプション	処理時間間隔文字列（例： '5 seconds' 、 '1 minute' ）。処理時間に基づいて、マイクロバッチクエリを定期的に実行します。
once	ブール値、オプション	True の場合、データのバッチを 1 つだけ処理し、クエリーを終了します。Databricks Runtime 11.3 LTS以降では非推奨。代わりに availableNow=True を使用してください。
continuous	文字列、オプション	Databricksではサポートされていません。このパラメーターは、連続パイプラインモードとは異なる Apache Spark の実験的な連続処理モードを有効にします。低遅延のStructured Streaming クエリーにはリアルタイムモードを使用します。
availableNow	ブール値、オプション	Trueの場合、利用可能なすべてのデータを複数のバッチで処理してからクエリを終了します。
realTime	文字列、オプション	バッチの期間を表す文字列。例：'5 minutes'。長時間実行されるバッチ内でレコードが到着したときに処理します。この値はバッチの期間を制御するものであり、レコード処理の待機時間を制御するものではありません。Structured Streaming のリアルタイムモードを参照してください。

戻り値

DataStreamWriter

例

Python

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

5秒ごとに実行をトリガーする：

Python

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

利用可能なすべてのデータを複数のバッチで処理する：

Python

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

5分間のバッチでリアルタイムモードを構成します。各バッチの実行中に、レコードは到着次第処理されます。サポートされているソース、シンク、コンピュートタイプを使用します：

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 ...>