Handledning: Bygg en filbehandlingspipeline med FILE-typen

Lär dig hur du bygger en medallionpipeline med Lakeflow-pipeline som behandlar ostrukturerade dokument från början till slut. Detta exempel använder samples.sec.contracts exempeldatasetet, en samling SEC-inlämnade juridiska avtal lagrade som PDF-filer i en Unity-katalogvolym.

Pipelinen tar in PDF:erna som hanterade FILE referenser med Auto Loader, tolkar varje dokument med AI-funktioner, klassificerar det som en avtalstyp och extraherar strukturerade fält för varje typ.

För typreferensen, se FILE typ.

I den här handledningen kommer du att:

• Mata in kontrakts-PDF:er från en volym som hanterade FILE-referenser stegvis med Auto Loader.
• Tolka varje dokument med ai_parse_document funktion och klassificera det med ai_classify funktion.
• Extrahera strukturerade fält för varje avtalstyp med ai_extract funktion.

Resultatet är en medaljongliknande pipeline: brons (råa hanterade FILE referenser), silver (parsade och klassificerade dokument) och guld (utvunna fält per avtalstyp). För mer information, se Vad innebär arkitekturen med medallion lakehouse? Bronslagret är en strömningstabell som inkrementellvis tar in filer, och silver- och guldlagren är materialiserade vyer som bara räknas om när deras indata ändras.

Requirements

För att slutföra den här självstudien måste du uppfylla följande krav:

• Var inloggad på en Azure Databricks-arbetsyta med Unity Catalog aktiverat.
• Ha FILE typen aktiverad för din arbetsyta. Workspace-administratörer kan aktivera det från sidan Förhandsvisningar . Se Hantera förhandsversioner av Azure Databricks.
• Ha behörigheter för att skapa tabeller i ett schema och för att skapa en pipeline.
• Ha en Unity Catalog-volym som du kan skriva till. Du deklarerar denna volym som bronstabellens FileSpace, och Unity Catalog kopierar de intagna filerna till den som hanterad lagring.
• Använd förhandsvisningskanalen.

Datasetet samples.sec.contracts är tillgängligt i alla arbetsytor som standard. Denna handledning lagrar de inhämtade PDF:erna som FILE MANAGED referenser: Unity Catalog kopierar varje fil till den volym du deklarerar som tabellens FileSpace och hanterar den tillsammans med tabellen, så när rader tas bort blir de refererade filerna berättigade till sopinsamling och tabellen och dess filer förblir synkroniserade. För att anpassa pipelinen till dina egna PDF:er, peka källsökvägen mot en volym som innehåller dina filer. För andra inmatningsalternativ, se Inmata filer som FILTYP.

Skapa filbehandlingspipelinen

Pipelinen bearbetar dokument i tre steg.

Steg 1. Brons: ta in råa PDF:er som hanterade filreferenser

Använd Auto Loader för att gradvis läsa kontrakts-PDF:erna från volymen. Att läsa filer med format => 'file' fångar en referens och metadata för varje fil utan att materialisera dess bytes. Att deklarera kolumnen som FILE MANAGED kopierar varje fil till tabellens FileSpace, volymen du sätter med databricks.filespace-preview tabellegenskapen, så Unity Catalog hanterar filerna med tabellen.

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/")
  )

• Fungerar för stora filer: en stor PDF finns i tabellens FileSpace, medan tabellraden endast lagrar en lättviktig FILE referens (uri, size, content_type, checksum). Jämför detta med BINARY typen, som inlinerar bytena i raden.
• Hanterad fillivscykel: Unity Catalog kopierar varje inhämtad fil till tabellens FileSpace och hanterar den med tabellen: att ta bort rader gör de refererade filerna berättigade till skräpsamling, så tabellen och dess filer förblir synkroniserade. För detaljer, se FILE MANAGED och FILE EXTERNAL.
• Inkrementell bearbetning: strömningstabellen tar stegvis in nya filer när de anländer till källan, utan att bearbeta befintliga filer igen. Datamängden samples.sec.contracts i detta exempel är statisk, men med en levande källa plockas nya filer upp vid varje pipelineuppdatering. För att också sprida källändringar och raderingar, ta in ändringsflödet med AUTO CDC. Se Tillämpa uppdateringar och raderingar med AUTO CDC.

Steg 2. Silver: tolka och klassificera dokument

Skicka varje FILE till ai_parse_document funktionen för att konvertera den råa PDF-filen till en strukturerad VARIANT som innehåller dokumentelement, layoutmetadata och text. Eftersom ai_parse_document accepterar en FILE-kolumn läser den dokumentet direkt från lagringsmediet och läser aldrig in bytena i klusterminnet.

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")
  )

Därefter skickar du den parsade utdatan till ai_classify funktionen för att tilldela varje dokument en av fem avtalstyper. Dokument med parsningsfel filtreras bort före klassificering. Detta exempel låser ai_classify till version 2.1, vilket innebär att klassificeringen returneras som ett objekt per etikett, så läs etiketten från nyckeln 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""")
  )

Steg 3. Gold: extrahera fält per avtalstyp

Varje avtalstyp har sin egen uppsättning relevanta fält. Filtrera de klassificerade dokumenten till en enda typ, skicka det tolkade innehållet till funktionen ai_extract med ett schema för de fält du vill ha och omvandla sedan svaret till typade kolumner. Detta exempel fäster ai_extract till version 2.1, där varje extraherat fält är ett objekt, så läs dess value nyckel.

Följande exempel bygger upp guldtabellen för konsultavtal:

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")
  )

Med dessa satser har du en helt inkrementell pipeline: när nya kontrakts-PDF:er anländer till volymen tar Auto Loader in dem som hanterade FILE referenser ai_parse_document och ai_classify routar varje dokument, och den consulting_agreements materialiserade vyn gold visar de extraherade fälten.

Exempelanteckningsböcker

Följande notebooks innehåller hela pipelinen från denna handledning. Dessa notebooks är pipeline-källkod, inte körbara notebooks. Importera anteckningsboken för ditt språk och ange sedan dess sökväg i fältet Source code när du konfigurerar pipelinen. Se Konfigurera arbetsflöden.

SQL

SQL-notebook för pipeline för filbearbetning

Hämta anteckningsbok

Python

Filbehandlingspipeline Python-notebook

Hämta anteckningsbok

Utforska på egen hand

Pipelinen klassificerar dokument i fem avtalstyper men extraherar endast fält för consulting_agreement. För att utöka den, upprepa guldsteget för varje återstående typ, ändra filtret contract_type och schemat ai_extract för att matcha fälten som är relevanta för den typen. Ett exempel:

• affiliate_agreement: party_1_name, party_2_name, , commission_ratepayment_frequency
• marketing_agreement: party_1_name, party_2_name, , effective_dateterritory
• hosting_agreement: provider_name, customer_name, , effective_dateterm_length
• escrow_agreement: owner_name, licensee_name, , escrow_agent_namesoftware_name

Ytterligare resurser

• FILE typ
• Inmata filer som FILE-typen
• FILE-funktionen snabbstart
• Läs mer om Auto Loader. Se Vad är en automatisk inläsare?.