Apache Spark Structured Streaming は、データを段階的に処理します。 トリガー間隔は、構造化ストリーミングが新しいデータをチェックする頻度を制御します。 ほぼリアルタイムの処理、スケジュールされたデータベースの更新、1 日または 1 週間のすべての新しいデータのバッチ処理のトリガー間隔を構成できます。
自動ローダーとは、構造化ストリーミングを使用してデータを読み込むため、トリガーのしくみを理解することで、目的の頻度でデータを取り込みながらコストを制御する最大の柔軟性を提供します。
重要
Azure Databricksでは、ユース ケースの待機時間とコストのバランスを取るトリガー モードを設定することをお勧めします。 そうしないと、クラウド プロバイダーから予期しないストレージ コストが発生する可能性があります。 詳細については、 クラウド ストレージ コストの制御 に関するページを参照してください。
トリガー モードの概要
次の表は、構造化ストリーミングで使用できるトリガー モードをまとめたものです。
| トリガー モード | 構文の例 (Python) | 最適な対象者 |
|---|---|---|
| 未指定 (既定値) | N/A | 待機時間が 3 ~ 5 秒の汎用ストリーミング。 0 ミリ秒間隔の processingTime トリガーと同じです。 ストリーム処理は、新しいデータが到着する限り継続的に実行されます。 |
| 処理時間 | .trigger(processingTime='10 seconds') |
コストとパフォーマンスのバランスを取る。 システムでデータのチェックが頻繁に行われるのを防ぐことで、オーバーヘッドを軽減します。 |
| 今すぐ利用可能 | .trigger(availableNow=True) |
スケジュールされた増分バッチ処理。 ストリーミング ジョブがトリガーされた時点で使用可能な量のデータを処理します。 |
| リアルタイム モード | .trigger(realTime='5 minutes') |
不正行為の検出やリアルタイムのパーソナル化など、2 秒未満の処理を必要とする超低待機時間の運用ワークロード。 パブリック プレビュー。 '5 分' はマイクロバッチの長さを示します。 クエリのコンパイルなどのバッチごとのオーバーヘッドを最小限に抑えるには、5 分を使用します。 |
| 継続的 | .trigger(continuous='1 second') |
サポートされていません。 これは、Spark OSS に含まれる試験的な機能です。 代わりにリアルタイム モードを使用してください。 |
:::note サーバーレス コンピューティング
サーバーレス コンピューティングでは、 Trigger.AvailableNow() と Trigger.Once() のみがサポートされます。 Databricks では Trigger.AvailableNow() を推奨しています。
サーバーレス コンピューティングでの継続的ストリーミングには、継続的パイプライン モードを使用します。トリガーされたパイプライン モードではなく、連続モードで行ってください。
ストリーミングの制限事項を参照してください。
:::
processingTime: 時間ベースのトリガー間隔
構造化ストリーミングは、時間ベースのトリガー間隔を "固定間隔マイクロバッチ" と言います。
processingTime キーワードを使用して、.trigger(processingTime='10 seconds') のように、期間を文字列として指定します。
この間隔の構成によって、新しいデータが到着したかどうかを確認するためにシステムがチェックを実行する頻度が決まります。 待機時間の要件とデータがソースに到達する速度のバランスを取るために処理時間を構成します。
AvailableNow: 増分バッチ処理
重要
Databricks Runtime 11.3 LTS 以降では、 Trigger.Once は非推奨です。 すべての増分バッチ処理ワークロードに対して Trigger.AvailableNow を使用します。
AvailableNow トリガー オプションでは、使用可能なすべてのレコードが増分バッチとして使用され、maxBytesPerTriggerなどのオプションを使用してバッチ サイズを構成できます。 サイズ設定オプションは、データ ソースによって異なります。
サポートされるデータ ソース
Azure Databricksでは、多くの構造化ストリーミング ソースからの増分バッチ処理に Trigger.AvailableNow を使用できます。 次の表に、各データ ソースに必要な Databricks Runtime の最小サポート バージョンを示します。
| ソース | Databricks の最低ランタイム バージョン |
|---|---|
| ファイル ソース (JSON、Parquet など) | 9.1 LTS |
| デルタ湖 | 10.4 LTS |
| 自動ローダー | 10.4 LTS |
| Apache Kafka | 10.4 LTS |
| キネシス | 13.1 |
OpenSharing (responseFormat=delta; responseFormat=parquet には delta-sharing-client 1.4.0 以降が必要) |
18.0 |
realTime: 超低待機時間の運用ワークロード
構造化ストリーミングのリアルタイム モードでは、末尾が 1 秒未満でエンドツーエンドの待機時間が実現され、一般的な場合は約 300 ミリ秒です。 リアルタイム モードを効果的に構成して使用する方法の詳細については、「 構造化ストリーミングのリアルタイム モード」を参照してください。
Apache Spark には、 継続的処理と呼ばれるトリガー間隔が追加されています。 このモードは、Spark 2.3 以降、試験段階として分類されています。 Azure Databricksでは、このモードはサポートも推奨もされません。 待機時間の短いユース ケースでは、代わりにリアルタイム モードを使用してください。
注
このページの連続処理モードは、 Spark 宣言パイプラインでの連続処理とは無関係です。
クラウド ストレージ コストを制御する
既定では、トリガー モードを設定しない場合、Structured Streaming によってトリガー モードが processingTime に設定され、間隔が 0に設定され、数ミリ秒ごとに新しいデータがチェックされます。 これにより、1 日あたり大量のクラウド ストレージ API 呼び出しが生成され、クラウド プロバイダーから予期しない料金が発生する可能性があります。
Azure Databricksでは、待機時間とコストの要件に適したトリガー モードを構成することをお勧めします。 時間ベースのトリガー間隔の構成については、 processingTime を参照してください。
実行間のトリガー間隔を変更する
同じチェックポイントを使用しながら、実行間のトリガー間隔を変更できます。
間隔を変更するときの動作
マイクロバッチの処理中に構造化ストリーミング クエリが停止した場合、そのマイクロバッチは、新しいトリガー間隔が適用される前に完了する必要があります。 トリガー間隔を変更すると、マイクロバッチが以前に指定した構成で処理される場合があります。 移行後の予期される動作を次に示します。
-
時間ベースの間隔から
AvailableNow: マイクロバッチは、使用可能なすべてのレコードが処理される前に段階的なバッチとして処理されることがあります。 -
AvailableNowから時間ベースの間隔: 最後のAvailableNowジョブがトリガーされたときに使用可能なすべてのレコードに対して処理が続行される場合があります。
クエリエラーからの復旧
増分バッチを使用してクエリエラーから復旧しようとすると、トリガー間隔の変更によって問題は解決されません。 Structured Streaming にはべき等マイクロバッチが必要であるため、前回の失敗したバッチを完了させる必要があります。 Apache Spark のフォールト トレランス セマンティクスを参照してください。
エラーを解決するには、ワーカー ノードのサイズを大きくするなど、コンピューティング容量をスケールアップします。 まれに、新しいチェックポイントを使用してストリームを再起動することが必要になる場合があります。