自動ローダーを使用してデータを取り込む

6 分

新しいデータファイルがクラウドストレージに継続的に到着する場合、取り込まれたファイルを手動で追跡することなく効率的に処理する方法が必要です。 自動ローダー は、新しいファイルが表示されたら自動的に検出して取り込むことで、この課題を解決します。状態やチェックポイントを手動で管理しなくても、1 回だけ保証できます。

自動ローダーのしくみを理解する

自動ローダーは、クラウドストレージの場所を監視し、新しいファイルを増分処理する構造化ストリーミングソースです。自動ローダーは、ディレクトリ全体を繰り返しスキャンするのではなく、既に処理されているファイルに関する状態を維持し、各ファイルが 1 回だけ取り込まれることを確認します。

自動ローダーストリームを開始すると、ディレクトリ内の既存のファイルを処理し、新しい到着を継続的に監視できます。ストリームはチェックポイントの場所に進行状況情報を格納します。これにより、中断された場合に停止した場所から再開できます。

自動ローダーは、次の 2 つのモードのいずれかを使用して新しいファイルを検出します。

ディレクトリ一覧モード: 自動ローダーは、新しいファイルを検出するために入力ディレクトリを定期的に一覧表示します。この方法では、ストレージアクセスを超える追加の構成は必要ありません。
ファイル通知モード: 自動ローダーは、クラウド通知サービスを使用して、新しいファイルが到着したときにイベントを受信します。このモードは、ディレクトリ一覧の繰り返しを回避するため、大規模なワークロードの方が効率的です。

ほとんどの運用ワークロードでは、Databricks では、Unity カタログの外部の場所でファイルイベントが有効になっているファイル通知モードをお勧めします。ファイルイベントを使用すると、自動ローダーはファイルが到着したときに直接通知を受け取ります。これにより、待機時間とクラウド API のコストが削減されます。

ファイルインジェスト用に自動ローダーを構成する

自動ローダーを使用してデータを取り込むには、cloudFilesでspark.readStream形式を使用します。次の例では、Azure Data Lake Storage から JSON ファイルを読み取り、Unity カタログテーブルに書き込みます。

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

cloudFiles.format オプションは、ソースファイルの形式を指定します。自動ローダーでは、JSON、CSV、Parquet、Avro、ORC、XML、テキスト、バイナリファイルがサポートされます。 cloudFiles.schemaLocationには、推論されたスキーマが格納されます。これにより、自動ローダーは、時間の経過に伴うスキーマの変更を追跡できます。スキーマの場所とチェックポイントの場所は同じでもかまいませんが、それらを個別に保持すると、推論されたスキーマを失うことなくストリームをリセットできます。

trigger(availableNow=True)設定では、現在使用可能なすべてのファイルが処理され、停止します。このパターンは、スケジュールされたバッチジョブに適しています。連続処理の場合は、トリガーを省略するか、 trigger(processingTime="1 minute") を使用してファイルを間隔で処理できます。

read_filesで SQL 構文を使用する

自動ローダーでは、 read_files テーブル値関数を使用した SQL 構文もサポートされます。この方法は、SQL を使用する場合、または SQL ノートブックで作業する場合に便利です。

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Lakeflow パイプラインでストリーミングテーブルで read_files を使用すると、自動ローダー機能が自動的に有効になります。増分処理を有効にするには、 STREAM キーワードを使用します。

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

このコンテキストでは、パイプラインによってチェックポイント処理とスキーマの進化が自動的に管理されます。

ヒント

Unity カタログを使用する場合は、チェックポイントとスキーマの場所をマネージドストレージの場所に格納します。これにより、一貫性のあるガバナンスが確保され、入れ子になったパスに関するアクセス許可の問題が回避されます。

スキーマの推論と進化を処理する

自動ローダーでは、ソースファイルのスキーマを自動的に検出できるため、スキーマを手動で定義する必要がなくなります。スキーマ推論が有効になっている場合、自動ローダーはファイルをサンプリングして列名とデータ型を決定します。

JSON ファイルと CSV ファイルの場合、自動ローダーはすべての列を既定で文字列として推論します。この保守的なアプローチにより、データ損失の原因となる可能性のある型の不一致を防ぐことができます。自動検出を有効にするには、 cloudFiles.inferColumnTypes オプションを設定します。

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

ソースデータに新しい列が導入されると、自動ローダーによって変更が検出され、ターゲットスキーマを自動的に進化させることができます。 cloudFiles.schemaEvolutionMode オプションは、この動作を制御します。

Mode	行動
`addNewColumns`	ストリームが失敗します。新しい列がスキーマに追加されます。再起動すると、ストリームは更新されたスキーマで続行されます。これは、スキーマが指定されていない場合の既定値です。
`addNewColumnsWithTypeWidening`	ストリームが失敗します。新しい列がスキーマに追加され、サポートされているデータ型の変更 (`INT`への`LONG`など) が自動的に拡大されます。サポートされていない型の変更は、 `_rescued_data`でキャプチャされます。 Databricks Runtime 16.4 以降が必要です。
`rescue`	スキーマは進化せず、ストリームは失敗しません。すべての新しい列が `_rescued_data` 列にキャプチャされます。
`failOnNewColumns`	ストリームが失敗します。指定されたスキーマが更新されるか、問題のあるデータファイルが削除されない限り、ストリームは再起動しません。
`none`	スキーマは進化せず、新しい列は無視され、 `rescuedDataColumn` オプションが設定されていない限り、データは復旧されません。これは、スキーマが指定されている場合の既定値です。

復旧されたデータ列 (_rescued_data) は、型の不一致や予期しない列など、予想されるスキーマと一致しないデータをキャプチャします。この機能により、ソースデータが期待に準拠していない場合にデータが失われるのを防ぐことができます。自動ローダーは、スキーマを推論するときにこの列を自動的に追加します。

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

復旧されたデータ列の名前を変更するには、 rescuedDataColumn オプションを使用します。

.option("rescuedDataColumn", "my_rescued_data")

特定の列に特定の型が必要であることがわかっている場合は、 スキーマヒント を使用して推論された型をオーバーライドします。

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

自動ローダーストリームの監視と最適化

運用ワークロードの場合、監視はインジェストの進行状況を追跡し、問題を特定するのに役立ちます。自動ローダーは、検出されたファイルの状態を照会する SQL 関数を提供します。

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

このクエリは、自動ローダーが検出したファイルに関するメタデータ (処理状態を含む) を返します。この情報を使用して、ファイルが想定どおりに取り込まれていることを確認できます。

自動ローダーは、次のような構造化ストリーミングの進行状況ダッシュボードを使用してメトリックを報告します。

numFilesOutstanding: ファイルは検出されましたが、まだ処理されていません
numBytesOutstanding: 処理を待機している合計バイト数

大量のワークロードの場合は、 cloudFiles.maxFilesPerTrigger または cloudFiles.maxBytesPerTriggerを使用して処理速度を制御します。これらの設定により、1 つのマイクロバッチが大きくなりすぎないようにします。

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Important

チェックポイントの場所には、重大な状態情報が格納されます。チェックポイントファイルを削除する可能性があるクラウドストレージライフサイクルポリシーは、ストリームの状態が破損し、最初から再起動する必要があるため、適用しないでください。

取り込まれたデータ内のファイルレベルのメタデータにアクセスするには、 _metadata 列を選択します。この非表示の列には、 file_path、 file_name、 file_size、 file_modification_timeなど、各ソースファイルに関する情報が含まれます。

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader は Lakeflow パイプラインとシームレスに連携し、チェックポイントの作成やスキーマ管理を自動的に行います。次のユニットでは、エンドツーエンドのデータインジェストワークフローに対して宣言型パイプライン定義内で自動ローダーを使用する方法について説明します。

フィードバック

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