Unity Catalog で自動ローダーを使用する

自動ローダーを使用すると、Unity Catalog によって構成された外部の場所からデータを安全に取り込むことができます。 Unity カタログを使用してストレージを安全に接続する方法の詳細については、 Unity カタログを使用したクラウド オブジェクト ストレージへの接続に関するページを参照してください。 自動ローダーでは、構造化ストリーミングを利用して増分処理を行います。推奨事項と制限事項については、「構造化ストリーミングで Unity Catalog を使用する」を参照してください。

Unity カタログの自動ローダー リソースの場所を指定する

Unity Catalog セキュリティ モデルでは、ワークロードで参照されているすべての保存場所が Unity Catalog によって管理されることを前提としています。 Databricks では、チェックポイントとスキーマの展開に関する情報は常に、Unity Catalog によって管理されている保存場所に格納することをお勧めしています。 Unity Catalog では、チェックポイントまたはスキーマの推論および展開のファイルをテーブル ディレクトリの下に入れ子にすることはできません。

Unity カタログを使用してクラウド ストレージからデータを取り込む

次の例では、実行中のユーザーが外部の場所に対する READ FILES アクセス許可、ターゲット テーブルに対する所有者特権、および次の構成と許可を持つものとします。

自動ローダーを使用して Unity カタログのマネージド テーブルに読み込む

次の例では、自動ローダーを使用して Unity カタログのマネージド テーブルにデータを取り込む方法を示します。

Python

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))

SQL

CREATE OR REFRESH STREAMING TABLE dev_catalog.dev_database.dev_table
AS SELECT * FROM STREAM read_files(
  'abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data',
  format => 'json'
);

Lakeflow パイプライン内のread_files ステートメントでCREATE STREAMING TABLEを使用すると、チェックポイントとスキーマの場所が自動的に管理されます。

自動ローダーを使用して Unity カタログの外部テーブルに読み込む

特定のストレージの場所にデータを保持するには、マネージド テーブルではなく Unity カタログ外部テーブルを使用します。 たとえば、外部テーブルを使用して、Databricks 以外のクライアントとデータを共有したり、既存のデータを登録したりします。 外部テーブルでは、ストレージ パスを設定します。 「外部テーブルを操作する」を参照してください。

Unity カタログの外部テーブルで自動ローダーを使用するには、最初にテーブルを CREATE TABLE ... LOCATION に登録してから、名前でストリーム配信します。 テーブルの場所は、アクセス許可を持つCREATE EXTERNAL TABLE内にある必要があります。 チェックポイントの場所は、Unity カタログで管理される外部の場所にも配置する必要があります。 テーブル データとは別のパスを使用します。

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"
table_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/external/dev_table"

# One-time: register the external table in UC.
spark.sql(f"""
  CREATE TABLE IF NOT EXISTS dev_catalog.dev_database.dev_table
  USING DELTA
  LOCATION '{table_path}'
""")

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))