メインコンテンツまでスキップ
に最終更新

本番運用ワークロードの Auto Loader を構成する

Auto Loader の設定に関する包括的なベスト プラクティスについては、ファイル検出モードの選択、スキーマ管理、データ品質の処理を含め、Auto Loader のベスト プラクティスを参照してください。

Databricks は、増分データ取り込みのためにLakeFlow PipelinesでAuto Loaderを使用することを推奨しています。LakeFlow PipelinesはApache Spark Structured Streamingの機能を拡張し、わずか数行の宣言型PythonまたはSQLを記述するだけで、本番運用品質のデータパイプラインをデプロイできます：

Databricks 、本番運用でAuto Loader実行するためのストリーミングのベスト プラクティスに従うこともお勧めします。 構造化ストリーミングに関する本番運用の考慮事項を参照してください。

注記

Lakeflow pipelines は、ほとんどの本番運用データ取り込みに Auto Loader を実行するための推奨される方法です。ワークロードに低レイテンシーの要件がなく、コンピュートコストの最小化が優先される場合は、代わりに Trigger.AvailableNow を使用する Trigger されるバッチジョブとして Auto Loader をスケジュールできます。コストに関する考慮事項を参照してください。

Auto Loaderのモニタリング​

以下のセクションでは、メトリクス、ログ、アラート、一般的なトラブルシューティングワークフローなど、本番運用でAuto Loaderをモニタリングする方法について説明します。ダッシュボードパターン、レイテンシー分析、スキーマドリフト検出を網羅した包括的なリファレンスについては、Auto Loaderのモニタリングと監視を参照してください。

Auto Loader によって検出されたファイルのクエリ​

Auto Loader は、ストリームの状態を検査するための SQL API を提供します。cloud_files_state関数を使用すると、 Auto Loaderストリームによって検出されたファイルに関するメタデータを見つけることができます。 cloud_files_stateをクエリして、 Auto Loaderストリームに関連付けられたチェックポイントの場所を提供します。

注記

cloud_files_state 関数は、Databricks Runtime 11.3 LTS 以降で使用できます。

SQL
SELECT * FROM cloud_files_state('path/to/checkpoint');

ストリームの更新を聞く​

Auto Loaderストリームをさらに監視するには、 Databricks Apache Sparkのストリーミング クエリ リスナー インターフェイスを使用することをお勧めします。 Databricksのモニタリング構造化ストリーミング クエリ」を参照してください。

Auto Loader は、バッチごとにストリーミング Query Listener にメトリクスを報告します。 バックログに存在するファイルの数と、 numFilesOutstanding および numBytesOutstanding メトリクスのバックログの大きさは、ストリーミング クエリ進行状況ダッシュボードの 生データ タブで確認できます。

JSON
{
  "sources": [
    {
      "description": "CloudFilesSource[/path/to/source]",
      "metrics": {
        "numFilesOutstanding": "238",
        "numBytesOutstanding": "163939124006"
      }
    }
  ]
}

Databricks Runtime 10.4 LTS以降でファイル通知モードを使用する場合、メトリクスには、 AWSおよびAzureのクラウド上でキューに入れられたファイル イベントのおおよその数もapproximateQueueSizeとして含まれます。

コストに関する考慮事項​

Auto Loader実行する場合、主なコストはコンピュート リソースとファイル検出です。

ワークロードに低レイテンシ要件がない場合、継続的に実行するのではなく、Lakeflow ジョブを使用して Auto Loader を Trigger.AvailableNow を使用してバッチジョブとしてスケジュールすることで、コンピュートコストを削減できます。構造化ストリーミングのトリガー間隔の設定を参照してください。これらのバッチジョブは、ファイルの到着から処理までの遅延をさらに低減するために、ファイル到着トリガーを使用してトリガーできます。

ファイルの検出コストは、ディレクトリリストモードでのストレージアカウントに対する LIST 操作、およびファイル通知モードでのサブスクリプションサービスとキューサービスに対する API リクエストという形で発生する可能性があります。Trigger.ProcessingTime などの連続 Trigger は、Auto Loader が新しいファイルを見つけるためにディレクトリ全体を継続的にリストするため、ディレクトリリストモードでは特にコストがかかります。ワークロードで連続 Trigger が必要な場合は、Databricks はレイテンシー要件に基づいてファイル検出モードを選択することをお勧めします。

  • 低遅延とシンプルさ ：ファイルイベントでAuto Loaderを使用します。 ファイルイベントでは、バケットごとにキューが1つだけ必要で、以降の実行では増分検出を使用します。詳細については、 「ファイルイベントを使用したAuto Loader概要」を参照してください。
  • レイテンシに非常に敏感なアプリケーション ：従来のファイル通知モードを使用してください。クラシックモードでは、ファイルイベントによって発生する追加のキャッシュホップなしに、クラウドキューから直接データを読み取ります。このモードでは、 Auto Loaderによって作成されたリソースにタグを付け、リソース タグを使用してコストを追跡できます。 詳細については、 「ファイル通知」を参照してください。

ソースデータの保持​

注記

Databricks Runtime 16.4 LTS 以降で利用できます。

ソースディレクトリにファイルが蓄積されるにつれて、ストレージコストが増加し、特にディレクトリ一覧表示モードではファイルの検出速度が低下します。Auto Loaderには、処理後にファイルをアーカイブまたは削除することでファイルの保持を自動的に管理するcloudFiles.cleanSourceオプションが用意されています。

コスト削減のため、ソースディレクトリ内のファイルをアーカイブする​

警告
  • cloudFiles.cleanSourceを設定すると、ソース ディレクトリ内のファイルが削除または移動されます。
  • データ処理にforeachBatch使用する場合、 foreachBatch操作が正常に返されるとすぐに、操作でバッチ内のファイルのサブセットのみが使用された場合でも、ファイルは移動または削除の候補になります。

Databricksは、検出コストを削減するために、ファイルイベントとAuto Loaderを併用することを推奨しています。これにより、検出が段階的に行われるため、コンピュートのコストも削減されます。

ファイル イベントを使用できず、ディレクトリ リストを使用してファイルを検出する必要がある場合は、 cloudFiles.cleanSourceオプションを使用して、Auto Loader がファイルを処理した後に自動的にアーカイブまたは削除し、検出コストを削減できます。Auto Loader は処理後にソース ディレクトリからファイルをクリーンアップするため、検出時にリストする必要があるファイルの数は少なくなります。

cloudFiles.cleanSource MOVEオプションとともに使用する場合、次の要件を考慮してください。

  • ソースディレクトリと移動先のディレクトリは、両方とも同じ外部ロケーション、ボリューム、またはDBFSマウント内にある必要があります。クロスバケットおよびクロスコンテナの移動はサポートされておらず、エラーになります。
  • 移動先はボリュームパス（例： /Volumes/my_catalog/my_schema/my_volume/archive/ ）にすることができます。
  • ソースと宛先ディレクトリが同じ外部ロケーションにある場合は、管理ストレージ (管理ボリュームやカタログなど) を含む兄弟ディレクトリを持たないでください。 このような場合、Auto Loader は宛先ディレクトリへの書き込みに必要な権限を取得できません。

Databricks では、次の場合にこのオプションを使用することをお勧めします。

  • ソース ディレクトリには、時間の経過とともに大量のファイルが蓄積されます。
  • コンプライアンスまたは監査のために、処理済みのファイルを保持する必要があります ( cloudFiles.cleanSourceをMOVEに設定)。
  • 取り込み後にファイルを削除することで、ストレージ コストを削減します ( cloudFiles.cleanSourceをDELETEに設定)。DELETEモードを使用する場合、 Databricks 、 Auto Loader削除がソフト削除として機能し、構成ミスが発生した場合に使用できるように、バケットのバージョン管理を有効にすることをお勧めします。 さらに、Databricks では、回復要件に基づいて、指定された猶予期間 (60 日または 90 日など) の経過後に古いソフト削除バージョンを消去するようにクラウド ライフサイクル ポリシーを設定することを推奨しています。

cleanSource オプションとそのdefaultに関する完全なリファレンスについては、「Clean up processed files with Auto Loader」を参照してください。

処理済みファイルをコールドストレージパスに移動​

次の例では、処理済みのファイルを14日後に同じバケット内のアーカイブディレクトリに移動するようにAuto Loaderを設定します。アーカイブパスにクラウドライフサイクルポリシーを適用することで、ファイルをより安価なストレージティア（例えば、AWS S3 Glacier、Azure Cool/Archive、またはGCS Coldline/Archive）に移行できます。

Python
# Step 1: Configure Auto Loader to move processed files to an archive path.
checkpoint = "/Volumes/my_catalog/my_schema/my_volume/checkpoints/ingest_stream"
archive_path = "s3://my-bucket/archive/landing/"

df = (spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.cleanSource", "MOVE")
  .option("cloudFiles.cleanSource.moveDestination", archive_path)
  .option("cloudFiles.cleanSource.retentionDuration", "14 days")
  .option("cloudFiles.schemaLocation", checkpoint)
  .load("s3://my-bucket/landing/")
)

# Step 2: Write to a Delta table.
(df.writeStream
  .option("checkpointLocation", checkpoint)
  .trigger(availableNow=True)
  .toTable("my_catalog.my_schema.raw_events")
)

# Step 3 (outside Databricks): Set up a cloud lifecycle policy on the
# archive path to transition files to cold storage after a grace period.
# For example, in AWS you can configure an S3 Lifecycle rule to move
# objects under s3://my-bucket/archive/landing/ to S3 Glacier after
# 30 days.

トリガーを使用する。今すぐ利用可能、レート制限あり​

注記

Databricks Runtime 10.4 LTS 以降で使用できます。

Auto Loader Trigger.AvailableNowを使用して、 LakeFlowジョブでバッチ ジョブとして実行するようにスケジュールできます。 AvailableNowトリガーは、クエリの開始時刻 前に 到着したすべてのファイルを処理するように Auto Loader に指示します。ストリームの開始後に到着した新しいファイルは、次のトリガーまで無視されます。

Trigger.AvailableNowを使用すると、ファイル検出はデータ処理と非同期に行われ、レート制限を使用して複数のマイクロバッチ間でデータを処理できます。Auto Loader by デフォルトは、マイクロバッチごとに最大 1000 個のファイルを処理します。 cloudFiles.maxFilesPerTrigger と cloudFiles.maxBytesPerTrigger を構成して、マイクロバッチで処理するファイルの数またはバイト数を構成できます。ファイル制限はハード制限ですが、バイト制限はソフト制限であり、指定された maxBytesPerTriggerよりも多くのバイトを処理できます。 両方のオプションが一緒に提供されると、 Auto Loader は制限の 1 つに達するために必要な数のファイルを処理します。

チェックポイントの場所​

チェックポイントの場所は、ストリームの状態と進行状況の情報を格納するために使用されます。Databricks では、チェックポイントの場所をクラウド オブジェクト ライフサイクル ポリシーのない場所に設定することをお勧めします。チェックポイントの場所にあるファイルがポリシーに従ってクリーンアップされた場合、ストリームの状態は破損します。これが発生した場合は、ストリームを最初から再起動する必要があります。

ファイルイベントの追跡​

Auto Loader keeps track of discovered files in the checkpoint location using RocksDB to guarantee that each file is ingested exactly one time. For high-volume or long-lived ingestion streams, Databricks recommends upgrading to Databricks Runtime 15.4 LTS or above. In these versions, Auto Loader does not wait for the entire RocksDB state to be downloaded before the stream starts, which can accelerate stream startup time. If you want to prevent the file states from growing without limits, you can also consider using the cloudFiles.maxFileAge option to expire file events that are older than a certain age. The minimum value that you can set for cloudFiles.maxFileAge is "14 days". Deletes in RocksDB appear as tombstone entries. Therefore, you might see the storage usage increase temporarily as events expire before it starts to level off.

警告

cloudFiles.maxFileAge は、大量のデータセットのコスト管理メカニズムとして提供されます。 cloudFiles.maxFileAgeを積極的に調整しすぎると、重複インジェストやファイルの欠落など、データ品質の問題が発生する可能性があります。したがって、 Databricks では、cloudFiles.maxFileAge に対して 90 日間などの保守的な設定を推奨します。これは、同等のデータ取り込みソリューションが推奨する設定と似ています。

cloudFiles.maxFileAgeオプションを調整しようとすると、未処理のファイルが Auto Loader によって無視されたり、既に処理済みのファイルの有効期限が切れて再処理されたりして、データが重複する可能性があります。 cloudFiles.maxFileAgeを選択する際に考慮すべき点は次のとおりです。

  • ストリームが長時間後に再起動すると、キューからプルされたファイル通知イベントのうち cloudFiles.maxFileAge より古いものは無視されます。 同様に、ディレクトリ一覧を使用する場合、ダウンタイム中に表示された可能性のある cloudFiles.maxFileAge より古いファイルは無視されます。
  • ディレクトリリストモードを使用し、 cloudFiles.maxFileAgeを使用する場合 (たとえば、 "1 month"に設定されている場合は、ストリームを停止し、 cloudFiles.maxFileAge を "2 months"に設定してストリームを再開します。1 か月以上経過し、2 か月以上経過したファイルは再処理されます。

If you set this option the first time you start the stream, you don't ingest data older than cloudFiles.maxFileAge. Therefore, if you want to ingest old data, don't set this option when you first start your stream. However, do set this option on subsequent runs.

cloudFiles.backfillInterval を使用して定期的なバックフィルをトリガーします。​

A backfill is an asynchronous directory listing that Auto Loader runs alongside normal file discovery to catch files that discovery missed. Even though cloud notification systems deliver events at least one time, a file can still be missed. A periodic backfill relists the source directory so that missed files are eventually discovered.

Set cloudFiles.backfillInterval to a duration string such as 1 day or 1 week to schedule periodic backfills. There is no default. In directory listing and classic file notification modes, backfills run only when you set this interval.

How backfills behave:

  • Time-based, not file-based : Auto Loader triggers the next backfill when the time since the last one exceeds the interval, tracking the last backfill time in the checkpoint rather than comparing file timestamps. To confirm the most recent backfill, use the lastBackfillStartTimeMs and lastBackfillFinishTimeMs metrics. See Monitoring Structured Streaming queries on Databricks.
  • Only missed files are ingested : A backfill ingests only files that weren't already processed, and new files continue to arrive through your configured discovery mode. It skips already-ingested files by checking the file state in the checkpoint, avoiding duplicates.
  • Asynchronous : Backfills run in the background and don't block micro-batch processing.

Set a backfill interval when you use classic file notification mode and have strict data completeness or service-level agreement (SLA) requirements. Auto Loader then relists the source on that cadence to catch missed notifications.

Don't set a backfill interval when using file events. Databricks automatically backfills those external locations with a full listing when file events are first enabled, and then continues to backfill about every 24 hours while a stream is ingesting. The interval setting is unsupported with file events, and because automatic backfill is less expensive, Databricks recommends using file events instead of setting a manual interval.

Each backfill is a full directory listing, so its cost scales with the number of files in the source directory and, in directory-listing mode, incurs LIST API charges. Choose the longest interval that still meets your completeness SLA.

Avoid full directory listings with file events​

When using file events, run your Auto Loader streams at least every 7 days to avoid a full directory listing. Running your Auto Loader streams this frequently ensures that file discovery is incremental.

包括的な管理対象ファイル イベントのベスト プラクティスについては、 「ファイル イベントを使用した Auto Loader のベスト プラクティス」を参照してください。

このページの見出し