新しいファイルが到着したときにジョブをトリガーする

ファイル到着トリガーを使用して、Amazon S3、Azure ストレージ、Google Cloud Storage などの外部の場所に新しいファイルが到着したときにジョブの実行をトリガーできます。この機能は、スケジュールされたジョブの効率が、不規則な新しいデータ到着によって損なわれる場合に便利です。

ファイル到着トリガーの仕組み

ファイル到着トリガーは、新しいファイルをベストエフォートで毎分チェックしますが、この動作は基盤となるクラウドストレージのパフォーマンスによって影響を受ける場合があります。ファイル到着トリガーには、ストレージロケーションのファイル一覧取得にかかるクラウドプロバイダーの料金以外、追加コストは発生しません。

ファイル到着トリガーは、Unity Catalog の外部ロケーションまたはボリュームのルート、あるいはそのサブパスを監視するよう構成できます。たとえば、Unity カタログボリューム /Volumes/mycatalog/myschema/myvolume/の場合、ファイル到着トリガーの有効なパスを次に示します。

/Volumes/mycatalog/myschema/myvolume/
/Volumes/mycatalog/myschema/myvolume/mydirectory/

トリガーは、構成したロケーションのすべてのサブディレクトリを再帰的にチェックし、新しいファイルの有無を確認します。たとえば、場所 /Volumes/mycatalog/myschema/myvolume/mydirectory/ のファイル到着トリガーを作成すると、この場所には次のサブディレクトリがあります。

/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirA
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirB
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirC/subdirD

トリガーは mydirectory、subdirA、subdirB、subdirC、subdirC/subdirD に新しいファイルがないかを確認します。

ファイルイベントを使用したファイル到着トリガー

パフォーマンスを最適化するために、ファイルイベントに対して外部の場所を有効にする必要があります。外部の場所に対してファイルイベントが有効になっている場合、Azure Databricksは内部サービスを使用して、クラウドプロバイダーからの変更通知を処理してインジェストメタデータを追跡します。このサービスでは、サービスによって決定されたローリングリテンション期間にわたって作成または更新された最新のファイルのメタデータが保持され、ファイル処理の効率が向上します。

外部の場所でファイルイベントを有効にしてから数分以内に、その外部の場所の対象となるパスを監視する既存のファイル到着トリガーがファイルイベントの有効化の恩恵を受け始め、新しいトリガーは数秒以内にメリットを得られます。

外部の場所でのファイルイベントのパフォーマンスと容量の利点の詳細については、「制限事項」を参照してください。ファイルイベントに関する一般的な質問については、「ファイルイベントに関する FAQ」を参照してください。

開始する前に

ファイル到着トリガーを使用するには、以下が必要です。

ワークスペースで Unity カタログが有効になっている必要があります。
ストレージの場所として、Unity Catalogで設定されたボリュームまたは外部の場所を使用する必要があります。「Unity カタログボリュームとは」と「外部の場所の概要」を参照してください。

Databricks では、マネージドファイルイベントの外部の場所を有効にすることをお勧めします。これらの外部の場所のボリュームは、既定でファイルイベントのサポートを受けます。ファイルイベントを有効にするには、外部の場所の所有者であるか、外部の場所に対する MANAGE 特権を持っている必要があります。ファイルイベントの利点については、「ファイルイベントを使用したファイル到着トリガー」を参照してください。
ストレージの場所に対する READ 権限と、ジョブに対する CAN MANAGE 権限が必要です。ジョブのアクセス許可については、「ジョブの ACL」を参照してください。

ファイル到着トリガーを追加する

ファイル到着トリガーをジョブに追加するには:

Azure Databricksワークスペースのサイドバーで、Jobs & Pipelinesをクリックします。
必要に応じて、ジョブ と 私が所有 フィルターを選択します。
ジョブの [名前] リンクをクリックします。
右側の [ジョブの詳細 ] ウィンドウで、[ トリガーの追加] をクリックします。
[トリガーの種類] で [ファイル到着] を選択します。
[ストレージロケーション] には、監視対象とする Unity Catalog の外部ロケーションまたはボリュームのルート、あるいはそのサブパスの URL を入力します。
（任意）実行がトリガーされる頻度を制御するには、詳細オプション（トリガー間の最小時間（秒）および最後の変更後の待機時間（秒））を設定します。セットアップの例については、「実行をトリガーする頻度の制御」を参照してください。
構成をテストするには、[テスト接続] をクリックします。
保存をクリックします。

このトリガーを後で編集、一時停止、または削除するには、[ジョブの詳細] ウィンドウの [スケジュールとトリガー] セクションを使用します。「既存のトリガーを管理する」を参照してください。

実行がトリガーされる頻度を制御する

ファイル到着トリガーの 2 つの詳細オプションは、ファイル到着をジョブ実行に変換する方法を制御します。これらのオプションは、 クールダウン と デバウンスという 2 つの一般的なレート制御パターンを適用します。

トリガー間の最小時間 (秒単位): この間隔 (実行間のクールダウン) ごとにジョブを最大 1 回の実行に制限します。実行が完了すると、クールダウン中に到着したファイルは、間隔が経過するまで新しい実行を開始しません。このオプションを使用すると、頻繁な到着によって実行が立て続けに作成されないように、実行の作成頻度の上限を設定できます。
最後の変更後の待機時間（秒）: 最新のファイルが到着した後、実行を開始する前にこの時間だけ待機し、新しいファイルが到着するたびにタイマーがリセットされます（デバウンス）。このオプションは、ファイルがバッチで到着し、すべてのファイルが到着した後にバッチ全体を 1 回の実行で処理する場合に使用します。

単独でオプションを設定することも、両方を一緒に設定することもできます。次の例を参照してください。

最大 15 分ごとに実行する

ファイルの到着時に実行を作成し、15 分ごとに実行する頻度を増やさない場合は、次の詳細オプションを設定します。

トリガー間の最小時間 (秒単位): 900

各実行が完了すると、ファイルが到着し続ける場合でも、トリガーは 900 秒 (15 分) 待機してから別の実行を開始します。これにより、実行の作成は 15 分あたり最大 1 回に制限されます。

バッチ全体が到着するまで待機します

ファイルがバッチで到着し、各バッチを 1 回の実行で処理する場合は、バッチ間のギャップよりも短く、バッチ内のファイル間のギャップよりも長い値に、 秒単位で最後の変更後に待機 を設定します。たとえば、新しいバッチが約 5 分ごとに開始される場合は、次の詳細オプションを設定します。

最後の変更後 (秒単位) を待ちます。 60

新しい各ファイルはタイマーをリセットするため、トリガーは 60 秒経過した後にのみ実行を開始し、新しい到着はありません。このセットアップでは、バッチ内のファイルが 60 秒以内に到着することを前提としているため、タイマーはバッチの途中で期限切れになりません。バッチは 60 秒以上離れているため、連続するバッチは 1 回の実行にマージされません。

頻度に上限を設け、完全なバッチがそろうまで待機

実行の作成頻度を上限にする場合と、バッチの途中で実行を開始しないようにする場合は、両方のオプションを組み合わせることができます。例えば次が挙げられます。

トリガー間の最小時間 (秒単位): 900
最後の変更後 (秒単位) を待ちます。 60

この構成では、トリガーはバッチのランディングが完了するまで (新しいファイルがない場合は 60 秒)、実行を開始するまで待機し、15 分ごとに複数の実行を開始しません。

到着時にファイルを検出して処理する

ファイル到着トリガーをトリガーしたファイルを処理するには、自動ローダーを使用できます。自動ローダーは、正確に 1 回の保証で新しいファイルを段階的かつ効率的に処理します。たとえば、次のスニペットを使用して、Delta テーブルにファイルを読み込みます。

このソリューションを使用するには、ファイル到着トリガーを使用してジョブを作成し、以下のコードを含むノートブックを追加します。各 [REPLACE] プレースホルダーを適切な値に置き換えます。

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.
sink_table = "[REPLACE]" # Delta table to write to

# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", checkpoint_location) \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .writeStream \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .toTable(sink_table)

カスタムロジックを使用して新しいファイルを処理する必要があり、新しいファイルの URL のみを検出する必要がある場合は、次のコードスニペットに示すように、代わりに foreachBatch を使用できます。 foreachBatchでは、少なくとも 1 回の処理が保証されることに注意してください。 foreachBatchの使用方法の詳細については、「foreachBatch を使用して任意のデータシンクに書き込む」を参照してください。

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.

def process_batch(batch_df, batch_id):
  file_url = batch_df.select("path").collect()[0].path
  # [REPLACE] Your custom function for processing newly arrived files


# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .drop("content") \
  .writeStream \
  .foreachBatch(process_batch) \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .start()

失敗したファイル到着トリガーの通知を受信する

ファイル到着トリガーの評価が失敗した場合に通知を受け取るには、ジョブ失敗時の電子メールまたはシステム宛先通知を構成します。「ジョブについての通知を追加する」をご覧ください。

制限

新しいファイルでのみ実行をトリガーします。既存のファイルを同じ名前のファイルで上書きしても、実行はトリガーされません。

ファイルイベントに対してストレージの場所が有効になっている場合、既存の空のファイルにコンテンツを追加すると、新しいファイル到着として処理されるため、実行がトリガーされます。
ファイルイベントは、 FlushWithClose イベントをリッスンしてファイルを処理します。 AZURE API の使用によっては、このイベントが生成されない場合があり、ファイルの検出が遅れる可能性があります。このシナリオを処理するには、クラシックファイル通知イベントに関するページを参照してください。

ファイル到着トリガーに使用されるパスには、外部テーブルやカタログとスキーマの管理された場所を含めてはなりません。
ファイル到着トリガーで使用するパスには、ワイルドカード (* や ? など) を含めることはできません。
ストレージの場所が Unity カタログの外部の場所として構成されていて、その外部の場所がファイルイベントに対して有効になっている場合:
- 保存場所のファイル数に制限はありません。
- 余分なファイルの更新が多すぎると、タイムアウト時にトリガーがエラーする可能性があります。
  
  Unity カタログの外部の場所またはボリュームのサブパスにファイル到着トリガーが設定されている場合、外部の場所のルートなど、そのサブパスの外部で変更を行うと、トリガーで処理する必要があるメタデータが増える可能性があります。変更の大きい環境では、トリガーが処理時間の制限を超え、エラー状態になる可能性があります。
  
  これを回避するには、監視するサブディレクトリに具体的にマップする Unity カタログボリュームを作成し、そのボリュームのルートでファイル到着トリガーを設定します。この方法では、ターゲットパスをトリガーの有効なルートとして分離し、関連のないルートレベルの変更を減らし、トリガーがエラー状態になるのを防ぎます。
- 既存のファイルが変更され、そのメタデータがローリングリテンション期間を過ぎると、その変更は新しいファイル到着として扱われ、ジョブの実行がトリガーされます。これを防ぐには、変更できないファイルのみを取り込むか、自動ローダーでファイル到着トリガーを使用してインジェストの進行状況を追跡できます。
ファイルイベントに対してストレージの場所が有効になっていない場合:
- Azure Databricks ワークスペース内のこのような場所で、ファイル到着トリガーを使用して最大 50 個のジョブを構成できます。
- ストレージの場所には、最大 10,000 個のファイルを含めることができます。構成したストレージロケーションが Unity Catalog の外部ロケーションまたはボリュームのサブパスである場合、10,000 ファイルの制限はサブパスに適用され、ストレージロケーションのルートには適用されません。たとえば、ストレージロケーションのルートではサブディレクトリを含めて 10,000 ファイルを超えても問題ありませんが、構成したサブディレクトリでは 10,000 ファイルを超えることはできません。

ファイルイベントの制限事項も参照してください。

S3 および GCS 外部ロケーションの存在しないパスでのファイル到着トリガー

構成されたディレクトリが存在しないか、Amazon S3 または Google Cloud Storage から削除された場合、ファイル到着トリガーは引き続きエラーなしで評価されます。この動作は、S3 と GCS の両方が存在しないディレクトリ、削除されたディレクトリ、空のディレクトリを区別しないために発生します。

その結果、存在しないディレクトリパスまたは削除されたディレクトリパスを監視するファイル到着トリガーが失敗したり、エラー通知が生成されたりすることはありません。トリガーは引き続き評価し、ファイルを見つけず、そのパスにファイルが再び追加されるまでジョブの実行をトリガーしません。これは予期される動作であり、エラー状態ではありません。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-06-29