はじめに

DatabricksにFILE型という新しいデータ型が追加されました。
非構造化ファイル（ドキュメント、画像、音声など）の参照をテーブルに格納し、ai_parse_documentなどのAI関数と組み合わせて処理できる機能です。

FILE型は2024年8月時点でBeta機能です。

これの何が嬉しいのかについては、公式blogで丁寧に解説されています。

また、既に解説記事もアップされています。早い！

個人的にFILE MANAGEDの挙動が気になっており、特にSharePointやGoogle Driveにあるドキュメントを同期的に取り込んで利用する際に非常に便利な気がしています。
と思っていたら公式ドキュメントにまさにGoogle Driveのファイルを取り込むチュートリアルが掲載されていました。

こちらをベースに、FILE MANAGED型を使ってGoogle Driveからファイルを増分取込し、処理を実行するパイプラインを作成してみます。

今回はDatabricksの製品版を利用しました。
現時点でFree Editionだと動作しませんでしたが、やがて利用できるようになるのではないかと思います。

FILE型とは

FILE型は、非構造化ファイル（PDF、画像、音声など）への参照をDeltaテーブルのカラムとして格納するためのデータ型です。

主な特徴は以下の通りです：

FILE EXTERNAL: ファイルをVolume上にそのまま残し、参照（URI、サイズ、content_type、checksum）だけをテーブルに保持
FILE MANAGED: ファイルの実体をFilespace Volume内にコピーして管理。外部ソース（Google Drive、SharePoint等）からの取り込みはこちらを使う
ai_parse_document等のAI関数と連携: FILEカラムを直接AI関数に渡してドキュメント解析が可能

今回はGoogle Driveからの取り込みなのでFILE MANAGEDを使います。

今回作るもの

以下のようなパイプライン構成を作ります。ユースケースとしてはGoogle Drive内の非構造データを利用して、RAG用のチャンクデータを作成するような想定です。また、Google Drive上のファイルが追加・更新・削除された場合、それに追随してチャンクデータも更新される挙動です。

Google Drive
    │
    ▼ (Auto Loader + readChangeFeed)
┌─────────────────────────────────┐
│ bronze_managed_files            │  ← Streaming Table (変更フィード取り込み)
└─────────────────────────────────┘
    │
    ▼ (Auto CDC / SCD Type 1)
┌─────────────────────────────────┐
│ silver_documents                │  ← Streaming Table (最新状態のみ保持)
└─────────────────────────────────┘
    │
    ▼ (ai_parse_document)
┌─────────────────────────────────┐
│ silver_parsed_documents         │  ← Materialized View (パース結果)
└─────────────────────────────────┘
    │
    ▼ (ai_prep_search)
┌─────────────────────────────────┐
│ silver_chunked_documents        │  ← Materialized View (チャンク結果)
└─────────────────────────────────┘

ポイントは以下です。

readChangeFeedを使ってファイルの追加・更新・削除を検知
Auto CDCで最新状態だけを保持するテーブルを作成
ai_parse_document → ai_prep_search でRAG用のチャンクまで一気通貫で処理

Step0. 前提・環境準備

必要なもの

Unity Catalog対応のDatabricksワークスペース
Google Drive用のUnity Catalog Connection（事前にセットアップ済みであること。説明は割愛）
Filespace用のUC Volume
パイプラインのチャネルをPREVIEWにしておくこと（FILE型のパイプラインサポートに必要）
パイプラインの設定値（Google Driveの接続名、URI、Filespace Volume）はパイプラインの「設定」に記載

Step1. Bronze: 変更フィードでファイルを取り込む

Bronzeレイヤーでは、Google DriveのファイルをreadChangeFeed付きで増分取り込みします。

transformations/bronze/bronze_managed_files.py

from pyspark import pipelines as dp

# 設定値を取得
GDRIVE_CONNECTOR = spark.conf.get("gdrive_connector")
GDRIVE_URI = spark.conf.get("gdrive_uri")
FILESPACE_VOLUME = spark.conf.get("filespace_volume")


@dp.table(
    name="bronze_managed_files",
    schema="_file_id STRING, _sequence BIGINT, _is_deleted BOOLEAN, path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED, org_file_name STRING",
    table_properties={"databricks.filespace-preview": FILESPACE_VOLUME},
)
def bronze_managed_files():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "file")
        .option("databricks.connection", GDRIVE_CONNECTOR)
        .option("cloudFiles.readChangeFeed", "true")
        .load(GDRIVE_URI)
        .selectExpr(
            "*",
            "_metadata.file_name as org_file_name",
        )
    )

readChangeFeedをtrueにすることで、通常のファイル列に加えて以下のメタデータ列が追加されます：

カラム	説明
`_file_id`	ファイルの一意識別子
`_sequence`	変更のシーケンス番号
`_is_deleted`	ファイルが削除されたかどうか

これらを使って、後段のAuto CDCで最新状態を管理します。

Step2. Silver: Auto CDCで最新状態を維持する

Bronzeテーブルには変更履歴が全て残りますが、下流で処理したいのは「現時点で存在するファイル」だけです。Auto CDCを使って、SCD Type 1で最新状態のみを保持するテーブルを作ります。

transformations/silver/silver_documents.py

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

FILESPACE_VOLUME = spark.conf.get("filespace_volume")


dp.create_streaming_table(
  name="silver_documents",
  comment="This table contains the documents ingested from the bronze table",
  table_properties={"databricks.filespace-preview" : FILESPACE_VOLUME}
)

dp.create_auto_cdc_flow(
  target = "silver_documents",
  source = "bronze_managed_files",
  keys = ["_file_id"],
  sequence_by = col("_sequence"),
  apply_as_deletes = expr("_is_deleted = true"),
  except_column_list = ["_is_deleted", "_sequence"],
  stored_as_scd_type = 1
)

これにより、

ファイルが追加されるとsilver_documentsに行が追加される
ファイルが更新されると該当行が上書きされる
ファイルが削除されると該当行が消える（_is_deleted = trueで削除扱い）

結果として、silver_documentsには常に「Google Drive上に現存するファイル」だけが残ります。
Google Driveの場合、ゴミ箱に入れただけでは削除にならず、ゴミ箱からも抹消されて初めて削除になることに注意してください。

Step3. Silver: ドキュメント解析 & チャンキング

最後に、ai_parse_documentとai_prep_searchを使ってドキュメントの解析とチャンキングを行います。

3-1. ai_parse_documentでパース

transformations/silver/silver_parsed_documents.py

from pyspark import pipelines as dp

FILESPACE_VOLUME = spark.conf.get("filespace_volume")


@dp.materialized_view(
    name="silver_parsed_documents",
    comment="Parsed content for each document, produced by ai_parse_document.",
    table_properties={"databricks.filespace-preview" : FILESPACE_VOLUME}
)
def silver_parsed_documents():
    return spark.read.table("silver_documents").selectExpr(
        "*", "ai_parse_document(file) AS parsed"
    )

ai_parse_document(file)にFILEカラムをそのまま渡せるのが便利ですね。
PDFや画像など、対応するファイル形式を自動で解析してくれます。

3-2. ai_prep_searchでチャンキング

transformations/silver/silver_chunked_documents.py

from pyspark import pipelines as dp

FILESPACE_VOLUME = spark.conf.get("filespace_volume")


@dp.materialized_view(
    name="silver_chunked_documents",
    comment="Chunked content for each document, produced by ai_parse_document.",
    table_properties={"databricks.filespace-preview": FILESPACE_VOLUME},
)
def silver_chunked_documents():
    return (
        spark.read.table("silver_parsed_documents")
        .filter("is_variant_null(parsed:error_status)")
        .selectExpr("*", "ai_prep_search(parsed) AS chunk")
    )

パースに失敗したドキュメント（parsed:error_statusがnullでないもの）をフィルタしてからai_prep_searchに渡しています。出力にはRAGで使えるチャンク情報が含まれます。

ハマりポイント

検証中に遭遇したハマりポイントを共有します。

1. Google DriveのURIはルートディレクトリを指定する必要がある

cloudFiles.readChangeFeedをtrueに設定する場合、Google Driveのルートフォルダ（共有ドライブのルートやマイドライブのルート等）のURIを指定する必要があります。

サブフォルダのURIを指定すると、変更フィードの取得時にエラーになります。

# ❌ サブフォルダを指定するとreadChangeFeedでエラー
https://drive.google.com/drive/folders/123456_subfolder

# ✅ ルートフォルダを指定する
https://drive.google.com/drive/u/0/folders/0ajjdhrke

ルート以外のフォルダでreadChangeFeedを使いたい場合は、ひとまずルートを指定して取り込んだ上で、後段でパスベースのフィルタリングをかける形になるかと思います。

readChangeFeedを使わない（追加のみ・削除追従不要の）場合は、サブフォルダのURIでも問題ありません。

2. Bronzeテーブルのスキーマは明示的に指定する

Spark宣言型パイプライン上でのFILE型のカラムは、スキーマを明示的に指定しないとFILE EXTERNALとして推論されるようです（たぶん。挙動からの推測です）。
Google Driveなどの外部ソースからの取り込みではFILE MANAGED（ファイル実体をFilespace Volumeにコピーする方式）を使う必要があるため、スキーマ指定は必須です。

# ❌ スキーマ未指定 → FILE EXTERNALになってしまう（外部ソースでは動作しない）
@dp.table(name="bronze_managed_files")
def bronze_managed_files():
    ...

# ✅ スキーマを明示的に指定してFILE MANAGEDを宣言
@dp.table(
    name="bronze_managed_files",
    schema="_file_id STRING, _sequence BIGINT, _is_deleted BOOLEAN, path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED, org_file_name STRING",
    table_properties={"databricks.filespace-preview": FILESPACE_VOLUME},
)
def bronze_managed_files():
    ...

なお、table_propertiesにdatabricks.filespace-previewを指定するのもFILE MANAGEDを使うための要件です。このプロパティでFilespace用のVolumeパスを指定します。

まとめ

DatabricksのFILE型とreadChangeFeed、Auto CDCを組み合わせて、Google Driveのドキュメントを増分処理するパイプラインを作ってみました。

個人的には、変更フィードの取得からAuto CDCでの最新状態管理、ai_parse_document→ai_prep_searchでのRAG前処理まで、宣言的パイプラインの中で一気通貫に書けるのはかなり体験が良いと思います。FILE型のおかげでバイナリデータを直接テーブルに突っ込まなくてよいのも嬉しいポイントです。データサイズが大きいファイルが含まれると効率性がかなり重くなるので。

とはいえ、まだBeta機能ということもあり、スキーマ指定の必要性などドキュメントだけでは分かりにくいハマりポイントもありました。この辺りは今後改善されていくのではないかと思います。

FILE型は非構造化データを構造化データと同様に扱うために良いアプローチだと感じています。
パイプラインを組むエンジニアにとって重要な要素と思いますのでGAが待ち遠しいですね。
これからも引き続き検証を進めていきたいと思います。