自動ローダーのベスト プラクティス

このページでは、ユース ケースに合わせて、信頼性が高く、コスト効率が高く大規模に実行されるように自動ローダーを構成するために適用できるベスト プラクティスについて説明します。

これらのベスト プラクティスにより、運用上のオーバーヘッドが軽減され、完全なディレクトリ スキャンによる不要な LIST API コスト、スキーマ ドリフトによるサイレント データ損失、チェックポイント構成の誤りによるパイプラインの再起動など、運用環境での診断が困難な一般的な問題が回避されます。

運用構成の詳細については、 実稼働ワークロードの自動ローダーの構成に関するページを参照してください。 監視と可観測性については、「Auto Loader の監視と可観測性」を参照してください。

適切な実行フレームワークを選択する

ユース ケースに最適な実行フレームワークは、パイプラインに対して必要な制御の量と、管理する運用オーバーヘッドの量によって異なります。 ほとんどのユーザーと運用パイプラインでは、Lakeflow パイプラインを使用した自動ローダーが適しています。 ただし、最大制御とカスタマイズが必要な場合は、構造化ストリーミングで自動ローダーを使用します。 マネージド エクスペリエンスで最も簡単なセットアップを行うには、使用可能な場合はマネージド LakeFlow コネクタを使用します。

Lakeflow パイプラインは、自動スケーリング、データ品質チェック、スキーマ進化処理、およびイベント ログによる監視を使用して構造化ストリーミングを拡張します。 Databricks では、ほとんどの運用インジェスト ワークロードに対して Lakeflow パイプラインが推奨されます。

適切なスケジュール設定とトリガーの種類を選択する

ユース ケースに最適なスケジュール設定とトリガーの種類は、待機時間の要件とファイル到着パターンによって異なります。 ほとんどのユース ケースでは、Databricks では、ファイル イベントが有効になっているファイル到着トリガーをお勧めします。 これは、新しいファイルが到着したときにのみコンピューティングが実行されるため、低コストで低待機時間のインジェストを実現します。 3 つのトリガーの種類は、パイプラインを開始するタイミングと頻度が異なります。

  • 継続的: パイプラインは停止せずに実行されます。 継続的なコンピューティング コストが高くなるため、1 秒未満の待機時間が難しい要件である場合にのみ使用します。 ファイル イベントとペアリングします。
  • ファイル到着トリガー: 新しいファイルがソースの場所に到着すると、パイプラインが開始されます。 低から中の待機時間または不規則なファイル到着パターンに最適です。 ファイル イベントを有効にする必要があります。 「新しいファイルが到着したときにジョブをトリガーする」を参照してください。
  • スケジュール済み: パイプラインは時間ベースのスケジュール (たとえば、1 時間ごと) で実行されます。 レイテンシ要件が緩やかな場合（数分から数時間）に使用します。 ディレクトリ一覧で動作しますが、ファイル イベントでは、完全なディレクトリ スキャンを回避することで、スケジュールされたモードでもコストが削減されます。

バッチ スケジュールに Trigger.AvailableNow を使用する方法の詳細については、「 Trigger.AvailableNow とレート制限の使用」を参照してください。

適切なファイル検出モードを選択する

自動ローダーでは、セットアップの複雑さ、スケーラビリティ、コストのトレードオフが異なる 3 つのファイル検出モードがサポートされています。

Mode セットアップの複雑さ Scalability Cost いつ使用するか
ファイルイベント（推奨） 低 (1 回限りアクセス許可の設定) 1 時間あたり何百万ものファイル 最低 ほとんどのワークロードの既定値
クラシック ファイル通知 高 (21 以上のクラウド構成オプション) 1 時間あたり何百万ものファイル Medium ファイル イベントが使用できない場合
ディレクトリ一覧 なし ディレクトリ サイズによって制限される 最も高い (LIST API コスト) 小さいディレクトリ、1 回限りのバックフィル、またはセキュリティ ポリシーによってファイル イベントが妨げる場合

ファイル イベントは、ストリームごとに 1 つではなく、外部の場所ごとに 1 つのサブスクリプションとキューを使用して、クラウド ストレージ リソースを統合します。 パフォーマンスの違いは大規模に大きくなります。ディレクトリ一覧では、すべてのトリガーでソース ディレクトリ全体をスキャンする必要があるため、インジェスト時間はディレクトリ サイズと共に増加します。 ファイル イベントは新しいファイル通知を直接配信するため、ディレクトリ内のオブジェクトの数に関係なくインジェスト時間は低いままです。

ファイル イベントを有効にする

ファイル イベントには、1 回限りのクラウド アクセス許可付与と、マネージド ファイル イベント サービスを使用するように構成された外部の場所が必要です。 設定が完了すると、その外部の場所から読み取るすべての自動ローダー ストリームで、追加の構成なしでファイル イベントを使用できます。

  1. クラウド プロバイダー側で必要なクラウドアクセス許可を付与します。 要件はクラウド プロバイダーによって異なります。 外部の場所のファイル イベントを設定するを参照してください。

  2. cloudFiles.useManagedFileEventsを自動ローダー クエリでtrueに設定します。

    df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.useManagedFileEvents", "true")
  .load("/path/to/data/dir"))

    完全なセットアップ手順については、「 ファイル イベントを使用して自動ローダーに移行する」を参照してください。

ファイル イベントを使用できない場合

次の場合、ファイル イベントを使用できない場合があります。

  • 外部の場所は、ファイル イベントで構成されていません。
  • 組織のセキュリティ ポリシーでは、共有外部の場所でファイル イベントを有効にすることはできません。

このような場合は、 クラシック ファイル通知モード または ディレクトリ一覧モードを使用します。 ファイル検出モードの完全な比較については、「 自動ローダーのファイル検出モードの比較」を参照してください。

スキーマの進化を管理する

自動ローダーはスキーマを自動的に推論しますが、スキーマの進化を構成する方法は、データの完全性とパイプラインの安定性に影響します。 戦略を選択するには、次の表を使用します。

シナリオ レコメンデーション
スキーマが既知で固定されている 明示的なスキーマを指定する .schema()
スキーマが不明です。追加の変更が必要です schemaEvolutionMode: addNewColumns
スキーマが不明です。型の変更が必要です schemaEvolutionMode: addNewColumnsWithTypeWidening
厳格なスキーマ契約が必須 schemaEvolutionMode: failOnNewColumns
任意または予測できないスキーマ Variant型として取り込む

戦略を選択したら、次のプラクティスを適用して、スキーマの進化の動作を微調整します。

既知のフィールド型にスキーマ ヒントを使用する

cloudFiles.schemaHints オプションを使用して、事前に知っているフィールドに型を適用しますが、他のフィールドのスキーマ推論は引き続き許可します。

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "id long, amount double")
  .load("/path/to/data/dir"))

互換性のある型の変更に型拡大を使用する

addNewColumnsWithTypeWidening スキーマの進化モードでは、データをint列にルーティングする代わりに、互換性のある型 (たとえば、longから_rescued_data) が自動的に拡大されます。 これにより、単純型の昇格を処理する後処理ジョブが不要になります。

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load("/path/to/data/dir"))

予測不能なスキーマ向けに Variant 型で取り込む

データが特定のスキーマに準拠していない場合、またはスキーマが継続的に変更される場合は、データを Variant 型として取り込みます。

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "data")
  .load("/path/to/data/dir"))

Variant では、クエリ時にスキーマが読み取られますが、構造化列のクエリよりも効率が低くなります。 スキーマ推論と進化の完全なメカニズムについては、「 自動ローダーでスキーマの推論と進化を構成する」を参照してください。

不適切なデータとデータ品質を処理する

次のプラクティスは、ダウンストリーム レイヤーに反映される前に、不適切なデータを検出、キャプチャ、および分離するのに役立ちます。

_rescued_dataと_corrupt_recordを有効にする

自動ローダーには、正常に解析できないデータをキャプチャするための 2 つの列が用意されています。

  • _rescued_data は、現在のスキーマと一致しないフィールドをキャプチャします。 自動ローダーによって自動的に追加されます。
  • _corrupt_record は、まったく解析できない行をキャプチャします。 columnNameOfCorruptRecordを使用して有効にします。
df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "_corrupt_record string")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .load("/path/to/data/dir"))

Databricks では、破損したレコードを見逃す可能性のある競合状態を回避するために、columnNameOfCorruptRecordよりもbadRecordsPathすることをお勧めします。

監視に Lakeflow パイプラインの期待値を使用する

Lakeflow パイプラインの期待値を設定して、 _rescued_data と _corrupt_record が通常の条件下で NULL されていることを確認します。 NULL 以外の値は、スキーマの誤差またはデータの破損を示します。

import dlt

@dlt.table
@dlt.expect("no rescued data", "_rescued_data IS NULL")
@dlt.expect("no corrupt records", "_corrupt_record IS NULL")
def bronze_table():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", "_corrupt_record string")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .load("/path/to/data/dir"))

破損したデータを分離する

調査のために、解析不可能なデータを含む行を専用シンクに分離します。 これにより、破損したデータがダウンストリーム レイヤーに伝達されなくなります。

import dlt

@dlt.table
def corrupt_records_sink():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NOT NULL")

@dlt.view
def clean_table():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NULL")

ソース ファイルのメタデータを使用してデータに注釈を付ける

自動ローダー インジェスト クエリに _metadata 列を含めます。 少なくとも、 file_path と file_modification_timeをキャプチャします。 これにより、データの問題を特定のソース ファイルにトレースし、完全なファイル ライフサイクルの cloud_files_state() に対して結合することができます。

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/path/to/data/dir")
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

詳細については、「 ファイル メタデータ」列を参照してください。

コストとパフォーマンスを最適化する

自動ローダーの 3 つの主なコスト 要因であるクラウド LIST API 呼び出し、アイドル 状態のコンピューティング、および長期的なストレージの増加を削減するプラクティスを次に示します。

  • ファイル イベントを使用して LIST を最小限に抑えるAPI コスト: ファイル イベントは増分ファイル検出を提供するため、各実行で完全なディレクトリ一覧が不要になります。 これは、自動ローダーに最も影響を与える 1 つのコスト最適化です。

  • イベント ドリブン処理にファイル到着トリガーを使用する: ファイル到着トリガーは、新しいファイルが到着したときにのみパイプラインを開始するため、アイドル状態のコンピューティング料金は発生しません。 「新しいファイルが到着したときにジョブをトリガーする」を参照してください。

  • 処理されたファイルを cloudFiles.cleanSource でアーカイブする: cloudFiles.cleanSource を使用して、処理されたファイルを自動的に削除または移動します。 これにより、ストレージ コストと、有効期間の長いストリームのディレクトリ 一覧のコストの両方が削減されます。 詳細については、 コストを削減するためのソース ディレクトリ内のファイルのアーカイブを参照してください。

    • delete モードを使用して、インジェスト後にファイルを削除します。
    • move モードを使用して、コンプライアンスまたは監査のためにファイルを別の場所にアーカイブします。
    df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.cleanSource", "delete")
  .load("/path/to/data/dir"))

    Warning

    複数の自動ローダー ストリームまたは他のクライアントが同じソース ディレクトリから読み取る場合は、 cloudFiles.cleanSource を有効にしないでください。

  • パフォーマンスの向上を活用する: 最新の Databricks ランタイムにアップグレードするか、サーバーレス コンピューティングを使用して、最近の自動ローダーのパフォーマンス向上の恩恵を受けます。

チェックポイント管理

チェックポイントには、ストリームの進行状況とファイルの状態が格納されます。 チェックポイントを正しく構成または失うには、完全な再起動が必要であるため、重要なインフラストラクチャとして扱います。

  • チェックポイントの場所にクラウド オブジェクト ライフサイクル ポリシーを適用しないでください。 チェックポイント ファイルが削除された場合、ストリームの状態は破損しているため、最初から再起動する必要があります。
  • ストリームとソース ディレクトリごとに個別のチェックポイントを使用します。
  • 長期間存続する高ボリュームのストリームでは、状態の増大を抑えるために cloudFiles.maxFileAge を検討してください。 保守的な設定 (90 日以上推奨) を使用します。 この値を過度に厳しく設定すると、ファイルが対象ウィンドウの範囲外になった場合に、Auto Loader がすでに取り込んだファイルを再処理してしまうおそれがあります。

詳細については、「 ファイル イベントの追跡」を参照してください。

ファイル イベントで最適なファイル検出にボリュームを使用する

ファイル イベントのパフォーマンスを向上させるには、自動ローダーが読み込むパスまたはサブディレクトリごとに 外部ボリューム を作成します。 クラウド パス (/Volumes/catalog/schema/volume など) ではなく、ボリューム パス (s3://bucket/path など) を自動ローダーに提供します。 これにより、最適化されたデータ アクセス パターンを使用してファイルの検出が最適化されます。

ファイル イベントのベスト プラクティスの詳細については、「 ファイル イベントを使用した自動ローダーのベスト プラクティス」を参照してください。

  • Last updated on