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

ファイルを FILE タイプとして取り込む

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

FILE タイプは、非構造化ファイル（ドキュメント、画像、音声）への参照をテーブル内に保存し、クエリーを実行します。このページでは、ファイルを検出して FILE 参照として取り込み、新しいファイルが到着するたびに追加で取り込む方法を説明します。

FILEタイプの詳細については、FILEタイプを参照してください。非構造化データを取り込むためのアプローチの概要については、「FILEタイプと非構造化データ」を参照してください。

注記

FILE 列には定義された順序がありません。FILE 列は、パーティション列、クラスタリング列、または Z-Order キーとして使用できません。情報については、制限を参照してください。

ストレージモード​

FILE 参照は、次の2つのモードのいずれかで保存できます。

  • FILE EXTERNAL Unity Catalogボリューム内に既に存在するファイルを参照します。Databricksは、ボリューム外に保存されたファイルのFILE EXTERNAL参照の保存をサポートしていません。
  • FILE MANAGED Unity Catalog 管理ストレージにファイルのコピーを保存します。SharePoint、Google Drive、SFTP など、ボリューム外のソースからのファイルは、取り込みを行い、FILE MANAGED として保存する必要があります。

list_files を使用してファイルを検出する​

list_filesテーブル値関数を使用して、パスで利用可能なファイルを検出します。ファイルごとに1行が返され、その path、size、modification_time、および FILE リファレンスが含まれます：

SQL
SELECT * FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

SharePoint、Google Drive、SFTP など、Unity Catalog 接続を必要とするソース内のファイルを検出するには、connection パラメーターを追加します。

SQL
SELECT * FROM list_files('https://example.sharepoint.com/sites/my-site/', connection => 'my_sharepoint_connection');

list_files defaultでファイルを再帰的に検出します。詳細については、「list_filesテーブル値関数」を参照してください。

ファイルをFILE参照として取り込む​

ファイルの保存場所に基づいて、取り込み方法を選択します。Unity Catalog ボリューム内に既にあるファイルを参照するには、FILE EXTERNAL を使用します。外部ソースからファイルを取り込むには、それらを FILE MANAGED としてマネージドストレージにコピーします。

ボリュームファイルをFILE EXTERNALとして取り込む​

Unity Catalog ボリュームに既に存在するファイルを取り込むには、list_files を指定した CREATE TABLE AS SELECT (CTAS) ステートメントを使用します。これにより、コンテンツをコピーせずに各ファイルをその場で参照する FILE EXTERNAL 列を持つテーブルが作成されます。次の例では、ファイル名、メタデータ、および各ファイルの FILE 参照を含む documents テーブルを作成します：

SQL
CREATE TABLE documents AS
  SELECT _metadata.file_name, *
  FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

外部ソースファイルを FILE MANAGED として取り込む​

SharePoint、Google Drive、SFTPなどのソース内のファイルに対してFILE参照を生成するには、まずファイルを取り込み、FILE MANAGEDとして保存してください。ボリューム外に保存されたファイルに対しては、FILE EXTERNALはサポートされていません。

次の例では、SharePoint から FILE MANAGED テーブルにファイルを取り込みます。

SQL
CREATE TABLE managed_documents (
  file_name STRING,
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents
  SELECT _metadata.file_name, *
  FROM read_files(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file');

パイプラインを使用して新しいファイルを増分取り込みする​

新しいファイルが到着したときにそれを取り込むには、 STREAM read_files(..., format => 'file') でソースを読み取る Lakeflow パイプラインのストリーミングテーブルを使用します。各パイプラインの更新では、前回の更新後に追加されたファイルのみが処理されます。read_files および Spark宣言型パイプラインを参照してください。

Google Drive などのソースからファイルをインクリメンタルにストリームするには:

  1. パイプラインのチャンネルを PREVIEW に設定します。パイプラインで FILE 参照を取り込むには、PREVIEW チャンネルが必要です。
  2. 次のコードのように、STREAM read_files(..., format => 'file')を使用してソースを読み取るストリーミングテーブルを定義します。
SQL
CREATE STREAMING TABLE streaming_documents (
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
)
TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
AS SELECT *
  FROM STREAM read_files(
    'https://drive.google.com/drive/folders/my-folder-id',
    connection => 'my_gdrive_connection',
    format => 'file');

AUTO CDC を使用した更新と削除の適用​

ストリーミングインジェストは新しいファイルを追加しますが、ソースからの更新や削除はキャプチャしません。これらの変更を適用するには、AUTO CDCを使用してソースの変更フィードを読み取ります。

警告

Databricks では、次の例のように、まず変更データをマネージドテーブルに格納し、そのテーブルに AUTO CDC を適用することをお勧めします。AUTO CDC を STREAM read_files(..., readChangeFeed => true) に直接適用すると、ダウンストリームフローごとにソースの変更フィードが再読み込みされるため、処理コストが増加する可能性があります。

2つのステップでチェンジデータフィードを取り込みます。次の例では、SharePoint からチェンジデータフィードを取り込み、それを SCD タイプ 1 としてターゲットのストリーミングテーブルに適用します：

  1. 次のコードのように、マネージドファイルを使用して変更データをストリーミングテーブルに書き込みます。read_files で readChangeFeed => true を設定して変更フィードを返します。これには _file_id、_sequence、_is_deleted のメタデータ列が含まれます。
SQL
CREATE OR REFRESH STREAMING TABLE documents_changes (
  _file_id STRING,
  _sequence BIGINT,
  _is_deleted BOOLEAN,
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
)
TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
AS SELECT *
  FROM STREAM read_files(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file',
    readChangeFeed => true);
  1. 次のコードのように、AUTO CDC を使用してそのテーブルからターゲットストリーミングテーブルに変更を適用します。_file_id をキーとして、_sequence をシーケンス列として、_is_deleted を削除の識別に使用します。
SQL
CREATE OR REFRESH STREAMING TABLE documents
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

CREATE FLOW documents_cdc AS AUTO CDC INTO
  documents
FROM STREAM documents_changes
  KEYS (_file_id)
  APPLY AS DELETE WHEN _is_deleted = true
  SEQUENCE BY _sequence
  COLUMNS * EXCEPT (_is_deleted, _sequence)
  STORED AS SCD TYPE 1;

インラインバイナリデータを FILE 参照に変換する​

テーブルがすでにファイルコンテンツをインラインバイナリデータとして格納している場合は、create_file 関数を使用してそのデータをストレージに書き込み、FILE リファレンスを作成します。

次の例では、ユーザーが生成したテーブル raw_documents を使用します。これには name 列と、バイナリデータを保持する content 列が含まれます。

FILE EXTERNAL としてボリュームにバイナリデータを書き込む​

ファイルを外部ファイルとして Unity Catalog ボリュームに書き込むには、次のコードのように destination_path を create_file に渡します。

SQL
CREATE TABLE documents (name STRING, file FILE EXTERNAL) USING DELTA;

INSERT INTO documents (name, file)
  SELECT
    name,
    create_file(
      content => content,
      destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name
    )
  FROM raw_documents;

バイナリデータを FILE MANAGED としてマネージドストレージに書き込む​

代わりにファイルをマネージドファイルとして保存するには、バイナリコンテンツのみを指定して create_file を呼び出します。destination_path を省略した場合、Unity Catalog はコンテンツをマネージドストレージロケーションにuploadします：

SQL
CREATE TABLE managed_documents (name STRING, file FILE MANAGED) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents (name, file)
  SELECT name, create_file(content => content)
  FROM raw_documents;

次のステップ​

このページの見出し