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

チュートリアル: FILEタイプを使用したファイル処理パイプラインの構築

備考

ベータ版

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

非構造化ドキュメントをエンドツーエンドで処理するLakeflowパイプラインを使用して、メダリオンパイプラインを構築する方法を学びます。この例では、Unity CatalogボリュームにPDFとして保存されたSEC提出の法的合意書のコレクションである、samples.sec.contractsサンプルデータセットを使用します。

このパイプラインは、Auto Loaderを使用してPDFをマネージドFILE参照として取り込み、AI関数で各ドキュメントを解析し、契約タイプに分類して、タイプごとに構造化フィールドを抽出します。

タイプのリファレンスについては、FILEタイプを参照してください。

このチュートリアルでは、次の方法を学習します。

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

結果はメダリオン形式のパイプラインとなります：ブロンズ（生の管理されたFILE参照）、シルバー（解析および分類されたドキュメント）、ゴールド（契約タイプごとに抽出されたフィールド）。See What is the medallion lakehouse architecture?情報については、こちらをご覧ください。ブロンズレイヤーはファイルをインクリメンタルに取り込む ストリーミングテーブル であり、シルバーレイヤーとゴールドレイヤーは入力が変更されたときにのみ再計算される マテリアライズドビュー です。

要件​

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

  • Unity Catalogが有効になっているDatabricksワークスペースにLogインしている必要があります。
  • ワークスペースで FILE タイプを有効にします。ワークスペース管理者は、 プレビュー ページからこの機能を有効にできます。Databricksのプレビューを管理するを参照してください。
  • スキーマにテーブルを作成し、パイプラインを作成する権限があること。
  • 書き込み可能な Unity Catalog ボリュームを用意します。このボリュームをブロンズテーブルの FileSpace として宣言すると、Unity Catalog は取り込まれたファイルをマネージドストレージとしてそのボリュームにコピーします。
  • プレビューチャンネルを使用します。

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

ファイル処理パイプラインを作成する​

パイプラインは、3つのステージでドキュメントを処理します。

ステップ 1. ブロンズ：生のPDFを管理されたFILE参照として取り込む​

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');
  • 大きなファイルに対応 : 大きな PDF はテーブルの FileSpace に配置され、テーブルの行には軽量な FILE 参照 (uri、size、content_type、checksum) のみが格納されます。これと、行内にバイトをインライン化する BINARY タイプを比較してください。
  • 管理されたファイルライフサイクル : Unity Catalog は、取り込まれた各ファイルをテーブルの FileSpace にコピーし、テーブルとともに管理します。行を削除すると、参照されているファイルがガベージコレクションの対象となるため、テーブルとそのファイルは同期された状態に保たれます。詳細については、FILE MANAGED および FILE EXTERNAL を参照してください。
  • インクリメンタル処理 : ストリーミングテーブルは、ソースに新しいファイルが到着するたびに、既存のファイルを再処理することなく、新しいファイルを段階的に取り込みます。この例の samples.sec.contracts データセットは静的ですが、ライブソースを使用すると、パイプラインが更新されるたびに新しいファイルが取得されます。ソースの変更と削除も伝播するには、AUTO CDC を使用して変更フィードを取り込みます。AUTO CDC を使用した更新と削除の適用を参照してください。

ステップ2. Silver：ドキュメントの解析と分類​

各 FILE を ai_parse_document 関数に渡して、生の PDF をドキュメント要素、Layoutメタデータ、およびテキストを含む構造化された 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;
注記

raw_contracts ストリーミングテーブル上のマテリアライズドビューとして解析ステップを定義することで、コンピュートが増分化されます。各パイプラインの更新では、テーブル全体ではなく、前回の更新以降に追加されたファイルに対してのみ ai_parse_document がランされます。ai_parse_document は最もコストのかかるステップであるため、これにより、すでに処理済みのドキュメントを再解析することを回避できます。マテリアライズドビューの増分更新にはServerless コンピュートが必要です。パイプラインをServerlessでランしてください。See Spark宣言型パイプライン.

次に、解析された出力を ai_classify 関数に渡し、各ドキュメントに5つの契約タイプのいずれかを割り当てます。解析エラーが発生したドキュメントは、分類の前に除外されます。この例では 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);
ヒント

分類の精度を向上させるには、ラベルの説明と instructions オプションを ai_classify に追加します。ai_classify関数を参照してください。

ステップ 3. ゴールド: 契約タイプごとにフィールドを抽出する​

各契約タイプには、それぞれに関連するフィールドのセットがあります。分類されたドキュメントを1つのタイプにフィルタリングし、解析されたコンテンツを目的のフィールドのスキーマと共に ai_extract 関数に渡し、応答を型付き列にフラット化します。この例では ai_extract をバージョン 2.1 にピン留めします。このバージョンでは各抽出フィールドがオブジェクトであるため、その 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;

これらのステートメントにより、完全に増分的なパイプラインが実現します。新しい契約書PDFがボリュームに到着すると、Auto Loaderがそれらを管理対象のFILEリファレンスとして取り込み、ai_parse_documentとai_classifyが各ドキュメントをルーティングし、consulting_agreementsゴールドマテリアライズドビューが抽出されたフィールドを表示します。

ノートブックの例​

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

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

ノートブックを新しいタブで開くOpen in Databricks

自分で探索する​

このパイプラインはドキュメントを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

その他のリソース​

このページの見出し