チュートリアル:ファイル形式でファイル処理パイプラインを構築する

Important

この機能はベータ版です。使用するには、ワークスペース管理者がファイルタイプをプレビューページからオンにする必要があります。 Manage Azure Databricks プレビューを参照してください。

このチュートリアルでは、Lakeflowパイプラインを使って非構造化文書をエンドツーエンドで処理するメダリオンシステムを構築します。この例は、 samples.sec.contracts サンプルデータセットを使用しており、これはSECが提出した法的契約の集まりをUnityカタログのボリュームにPDFとして保存しています。

パイプラインはAuto LoaderでPDFを管理 FILE 参照として取り込み、AI関数で各文書を解析し、契約タイプに分類し、各タイプごとに構造化フィールドを抽出します。

型の参照については、 FILE 型を参照してください。

このチュートリアルでは、次のことを行います。

Auto Loaderでボリュームから契約PDFを管理された FILE 参照として段階的に取り込みます。
各文書を ai_parse_document 関数で解析し、 ai_classify 関数で分類します。
ai_extract関数を使用して、各契約タイプの構造化フィールドを抽出します。

その結果、メダリオン型のパイプラインが形成されます。ブロンズ(生の管理 FILE 参照)、シルバー(解析・分類された文書)、ゴールド(契約タイプごとの抽出フィールド)です。詳細については、「 medallion lakehouse のアーキテクチャとは」を参照してください。ブロンズレイヤーはファイルを段階的に取り込むストリーミングテーブルで、シルバー層とゴールド層は入力が変わった場合にのみ再計算されるマテリアル化されたビューです。

Requirements

このチュートリアルを完了するには、次の要件を満たす必要があります。

Unity Catalogを有効にしてAzure Databricksワークスペースにログインしてください。
ワークスペースで FILE を有効にしてください。ワークスペース管理者は プレビュー ページから有効にできます。 Manage Azure Databricks プレビューを参照してください。
スキーマ内でテーブルを作成し、パイプラインを作成する権限を持ちましょう。
書き込み可能なUnity Catalogボリュームがある。このボリュームをブロンズテーブルの FileSpaceとして指定すると、Unity Catalogが取り込んだファイルをマネージドストレージ内にコピーします。
プレビューチャンネルをご利用ください。

samples.sec.contractsデータセットはデフォルトですべてのワークスペースで利用可能です。このチュートリアルでは、取り込んだPDFを FILE MANAGED 参照として保存します。Unity Catalogは各ファイルをテーブルの FileSpace として宣言したボリュームにコピーし、テーブルで管理するため、行を削除すると参照ファイルがガベージコレクション対象となり、テーブルとそのファイルが同期されたままになります。自分のPDFにパイプラインを適応させるには、ソースパスをファイルを含むボリュームに向けてください。その他の取り込みオプションについては、「ファイル形式」として「ファイルを取り込み」を参照してください。

ファイル処理パイプラインの作成

パイプラインは文書を3段階で処理します。

ステップ 1. ブロンズ:管理されたファイル参照として生のPDFを取り込む

Auto Loaderを使って、巻から契約書のPDFを段階的に読み取ってください。 format => 'file'でファイルを読み取ることで、各ファイルの参照とメタデータを取得できますが、そのバイトは生成されません。列を FILE MANAGED として宣言すると、各ファイルがテーブルの FileSpace、つまり databricks.filespace-preview テーブルプロパティで設定したボリュームにコピーされるため、Unity Catalogがテーブルとともにファイルを管理します。

SQL

CREATE OR REFRESH STREAMING TABLE raw_contracts (
  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(
    '/Volumes/samples/sec/contracts/',
    format => 'file');

Python

from pyspark import pipelines as dp

@dp.table(
  name="raw_contracts",
  schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)
def raw_contracts():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .load("/Volumes/samples/sec/contracts/")
  )

大きなファイルには対応します。大きなPDFはテーブルの FileSpaceに存在し、テーブルの行には軽量な FILE 参照(uri、 size、 content_type、 checksum)のみが保存されます。これを BINARY 型と比較してください。型は行内のバイトをインライン化します。
管理されたファイルライフサイクル: Unity Catalogは取り込んだ各ファイルをテーブルの FileSpace にコピーし、テーブルで管理します。行を削除すると参照ファイルがガベージコレクション対象となり、テーブルとそのファイルが同期し続けます。詳細については FILE MANAGED and FILE EXTERNAL を参照してください。
増分処理:ストリーミングテーブルは、既存のファイルを再処理せずに、ソースに到着した新しいファイルを段階的に取り込みます。この例の samples.sec.contracts データセットは静的ですが、ライブソースの場合はパイプラインの更新ごとに新しいファイルが取得されます。ソースの変更や削除も伝播させるには、 AUTO CDCで変更フィードを取り込みます。 AUTO CDCでの更新および削除の適用を参照してください。

ステップ 2. シルバー:文書を解析し分類する

各FILEをai_parse_document関数に渡して、生のPDFを文書要素、レイアウトメタデータ、テキストを含む構造化されたVARIANTに変換します。 ai_parse_documentはFILE列を受け入れているため、ドキュメントをストレージから直接読み込み、バイトをクラスターメモリに読み込むことはありません。

SQL

CREATE OR REFRESH MATERIALIZED VIEW parsed_contracts AS
  SELECT
    path,
    ai_parse_document(file) AS parsed
  FROM raw_contracts;

Python

@dp.materialized_view(name="parsed_contracts")
def parsed_contracts():
  return (
    spark.read.table("raw_contracts")
      .selectExpr("path", "ai_parse_document(file) AS parsed")
  )

Note

解析ステップを raw_contracts ストリーミングテーブル上の物質化されたビューとして定義することで、計算が増分化されます。各パイプラインの更新は、前回の更新以降に追加されたファイルだけ ai_parse_document 実行され、テーブル全体には動作しません。 ai_parse_documentが最もコストがかかるため、すでに処理済みの文書を再解析する手間を避けられます。マテリアル化されたビューのインクリメンタルリフレッシュにはサーバーレス計算が必要です。パイプラインはサーバーレスで実行してください。 Spark 宣言型パイプラインに関するページを参照してください。

次に、解析された出力を関数に渡し、各文書に5つの合意タイプのいずれかを割り当てますai_classify。解析エラーが発生したドキュメントは、分類の前に除外されます。この例は ai_classify バージョン2.1にピン留めされており、分類はラベルごとのオブジェクトとして返されるため、 value キーからラベルを読み取ってください。

SQL

CREATE OR REFRESH MATERIALIZED VIEW classified_contracts AS
  SELECT
    path,
    parsed,
    ai_classify(
      parsed,
      '["affiliate_agreement", "marketing_agreement", "consulting_agreement", "hosting_agreement", "escrow_agreement"]',
      map('version', '2.1')
    ):response[0].value::STRING AS contract_type
  FROM parsed_contracts
  WHERE is_variant_null(parsed:error_status);

Python

@dp.materialized_view(name="classified_contracts")
def classified_contracts():
  return (
    spark.read.table("parsed_contracts")
      .filter("is_variant_null(parsed:error_status)")
      .selectExpr(
        "path",
        "parsed",
        """ai_classify(
             parsed,
             '["affiliate_agreement", "marketing_agreement", "consulting_agreement", "hosting_agreement", "escrow_agreement"]',
             map('version', '2.1')
           ):response[0].value::STRING AS contract_type""")
  )

Tip

分類の正確性を高めるために、ラベルの説明とai_classifyのinstructionsオプションを追加してください。 ai_classify 関数を参照してください。

ステップ 3. ゴールド:合意タイプごとのフィールド抽出

各契約タイプには独自の関連フィールドがあります。分類済み文書を1つのタイプにフィルターし、解析した内容を望むフィールドのスキーマで ai_extract 関数に渡し、応答をタイプ化された列にフラット化します。この例はバージョン2.1に ai_extract ピンで固定されており、各抽出されたフィールドはオブジェクトなので、その value キーを読み取る。

以下の例は、コンサルティング契約のためのゴールドテーブルを構築します。

SQL

CREATE OR REFRESH MATERIALIZED VIEW consulting_agreements AS
  WITH extracted AS (
    SELECT
      path,
      ai_extract(
        parsed,
        '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
        map('version', '2.1')
      ) AS fields
    FROM classified_contracts
    WHERE contract_type = 'consulting_agreement'
  )
  SELECT
    path,
    fields:response.company_name.value::STRING AS company_name,
    fields:response.consultant_name.value::STRING AS consultant_name,
    fields:response.compensation_amount.value::STRING AS compensation_amount,
    fields:response.effective_date.value::STRING AS effective_date
  FROM extracted;

Python

@dp.materialized_view(name="consulting_agreements")
def consulting_agreements():
  return (
    spark.read.table("classified_contracts")
      .filter("contract_type = 'consulting_agreement'")
      .selectExpr(
        "path",
        """ai_extract(
             parsed,
             '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
             map('version', '2.1')
           ) AS fields""")
      .selectExpr(
        "path",
        "fields:response.company_name.value::STRING AS company_name",
        "fields:response.consultant_name.value::STRING AS consultant_name",
        "fields:response.compensation_amount.value::STRING AS compensation_amount",
        "fields:response.effective_date.value::STRING AS effective_date")
  )

これらのステートメントを使えば、完全なインクリメンタルパイプラインが実現します。新しい契約PDFがボリュームに届くと、Auto Loaderはそれらを管理された FILE 参照として取り込み、ai_parse_document と ai_classify が各ドキュメントをルーティングし、consulting_agreements ゴールドの物質化ビューが抽出されたフィールドを表示します。

サンプルノートブック

以下のノートブックには、このチュートリアルの完全なパイプラインが含まれています。これらのノートブックはパイプラインのソースコードであり、実行可能なノートブックではありません。言語用のノートブックをインポートし、パイプラインを設定する際に ソースコード 欄でそのパスを指定してください。「パイプラインの構成」を参照してください。

SQL

ファイル処理パイプラインSQLノートブック

ノートブックを入手

Python

ファイル処理パイプライン Pythonノートブック

ノートブックを入手

自分で探検してみてください

パイプラインは文書を5つの契約タイプに分類しますが、抽出するフィールドは consulting_agreementのみです。拡張するには、残った各タイプに対してゴールドステップを繰り返し、 contract_type フィルターと ai_extract スキーマをそのタイプに関連するフィールドに合わせて変更します。例えば次が挙げられます。

affiliate_agreement: party_1_name、 party_2_name、 commission_rate、 payment_frequency
marketing_agreement: party_1_name、 party_2_name、 effective_date、 territory
hosting_agreement: provider_name、 customer_name、 effective_date、 term_length
escrow_agreement: owner_name、 licensee_name、 escrow_agent_name、 software_name

その他のリソース

FILE 型
ファイルをファイル形式として取り込む
FILE 関数 quickstart
自動ローダーの詳細については、こちらを参照してください。「自動ローダーとは」を参照してください。

フィードバック

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

最終更新: 2026-10-09