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

チュートリアル: 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インしている必要があります。
  • スキーマにテーブルを作成し、パイプラインを作成する権限があること。
  • プレビューチャンネルを使用します。

samples.sec.contracts データセットはdefaultですべてのワークスペースで利用可能なため、追加のセットアップは不要です。ファイルはすでに Unity Catalog ボリューム内に存在するため、このチュートリアルでは、その内容をコピーせずに FILE EXTERNAL 参照として保存します。パイプラインを独自の PDF に適合させるには、ソースパスをファイルが含まれているボリュームに向けます。その他の取り込みオプションについては、「FILE 型としてファイルを取り込む」を参照してください。

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

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

ステップ 1. ブロンズ：生PDFを外部FILE参照として取り込む​

Auto Loader を使用して、ボリュームから契約書の PDF を段階的に読み取ります。format => 'file' を使用してファイルを読み取ると、バイトを具体化することなく、各ファイルの参照とメタデータが取得されます。列を FILE EXTERNAL として宣言すると、その内容をコピーすることなく、各ファイルをその場で参照します。

SQL
CREATE OR REFRESH STREAMING TABLE raw_contracts (
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE EXTERNAL
)
AS SELECT *
  FROM STREAM read_files(
    '/Volumes/samples/sec/contracts/',
    format => 'file');
  • 大きなファイルに対応 : 大きなPDFはボリューム内に保持され、テーブル行には軽量な FILE リファレンス (uri、size、content_type、checksum) のみが格納されます。これを、行内のバイトをインライン化する BINARY タイプと比較してください。
  • インクリメンタル処理 : ストリーミングテーブルは、ソースに新しいファイルが到着するたびに、既存のファイルを再処理することなく、新しいファイルを段階的に取り込みます。この例の 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 ゴールドマテリアライズドビューが抽出されたフィールドを表示します。

自分で探索する​

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

その他のリソース​

このページの見出し