構造化ストリーミングのトリガー間隔を構成する

Apache Spark 構造化ストリーミングはデータを段階的に処理します。バッチ処理のトリガー間隔を制御すると、ほぼリアルタイムの処理、5 分ごとまたは 1 時間に 1 回の更新、日または週のすべての新しいデータのバッチ処理などのワークロードに構造化ストリーミングを使用できます。

Databricks Auto Loader は構造化ストリーミングを使用してデータを読み込むため、トリガーのしくみを理解することで、目的の頻度でデータを取り込む際にコストを制御する柔軟性を最大限に高めることができます。

時間ベースのトリガー間隔の指定

構造化ストリーミングでは、時間ベースのトリガー間隔を「固定間隔マイクロバッチ」と呼びます。 processingTime キーワードを使用して、期間を文字列として指定します (.trigger(processingTime='10 seconds')など)。

指定した trigger 間隔が小さすぎる (数十秒未満) 場合、システムは新しいデータが到着するかどうかを確認するために不要なチェックを実行する場合があります。待機時間の要件とデータがソースに到着する速度のバランスを取るように処理時間を構成します。

増分バッチ処理の構成

重要

Databricks Runtime 11.3 LTS 以降では、 Trigger.Once 設定は非推奨です。Databricks では、すべての増分バッチ処理ワークロードに Trigger.AvailableNow を使用することをお勧めします。

available now トリガーオプションは、使用可能なすべてのレコードを増分バッチとして使用し、 maxBytesPerTrigger などのオプションを使用してバッチサイズを構成できます (サイズ設定オプションは Data によって異なります)。

Databricks では、多くの構造化ストリーミングソースからの増分バッチ処理に Trigger.AvailableNow を使用できます。次の表に、各 Databricks Runtime に必要なサポートされている最小バージョンを示します。

ソース	最小 Databricks Runtime バージョン
ファイルソース (JSON、 Parquetなど)	9.1 LTS
Delta Lake	10.4 LTS
オートローダー	10.4 LTS
Apache Kafka	10.4 LTS
Kinesis	13.1

デフォルトのトリガー間隔はいくつですか?

構造化ストリーミングデフォルトから500msの固定間隔のマイクロバッチ。 Databricks では、新しいデータが到着したかどうかを確認し、小さすぎるバッチの処理に関連するコストを最小限に抑えるために、常に調整された trigger を指定することをお勧めします。

実行間のトリガー間隔の変更

同じチェックポイントを使用しながら、実行間のトリガー間隔を変更できます。

マイクロバッチの処理中に構造化ストリーミングジョブが停止した場合、新しいトリガー間隔が適用される前にそのマイクロバッチを完了する必要があります。そのため、トリガー間隔を変更した後、以前に指定した設定でマイクロバッチ処理が発生する場合があります。

時間ベースの間隔から AvailableNowを使用する場合、使用可能なすべてのレコードをインクリメンタルバッチとして処理する前に、マイクロバッチ処理が発生する可能性があります。

AvailableNow 間隔から時間ベースの間隔に移行すると、最後の AvailableNow ジョブがトリガーされたときに使用可能だったすべてのレコードの処理が続行される可能性があります。これは予期される動作です。

注：

増分バッチに関連するクエリー障害から回復しようとしている場合、バッチを完了する必要があるため、トリガー間隔を変更してもこの問題は解決しません。 Databricks では、バッチの処理に使用されるコンピュート容量をスケールアップして、問題の解決を試みることをお勧めします。まれに、新しいチェックポイントでストリームを再起動する必要がある場合があります。

連続処理モードとは何ですか?

Apache Spark では、継続的処理と呼ばれる追加のトリガー間隔がサポートされています。このモードは、Spark 2.3 以降、実験的モードとして分類されています。Databricks アカウントチームに相談して、この処理モデルのトレードオフを理解していることを確認してください。

この連続処理モードは、 Delta Live Tablesで適用された連続処理にはまったく関係しないことに注意してください。