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 i 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:

• Inkrementell inmata kontrakts-PDF:er från en volym som hanterade FILE referenser 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 arbetsplats. 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 inhämtade 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 och FileSpace hanterar den med tabellen, så rader som rader gör att de refererade filerna är lämpliga för skräpsamling 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 FIL-referenser

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 , volymen FileSpacedu 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: rader rader gör de refererade filerna lämpliga för 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 FILEfunktionai_parse_document för att konvertera rå-PDF:en till en struktur VARIANT som innehåller dokumentelement, layoutmetadata och text. Eftersom ai_parse_document den accepterar en FILE kolumn läser den dokumentet direkt från lagringen och laddar aldrig 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 fäster ai_classify till version 2.1, som returnerar klassificeringen som ett per-etikettobjekt, 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. Guld: utvinna fält per avtalstyp

Varje avtalstyp har sin egen uppsättning relevanta fält. Filtrera de klassificerade dokumenten till en typ, skicka det parsade innehållet till ai_extract att fungera med ett schema av de fält du vill ha, och platta sedan ut svaret i 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 guldmaterialiserade vyn visar de extraherade fälten.

Exempelanteckningsböcker

Följande anteckningsböcker 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 källkodsfältet när du konfigurerar pipelinen. Se Konfigurera arbetsflöden.

SQL

SQL-notebook för filbearbetningspipeline

Hämta anteckningsbok

Python

Filbehandlingspipeline Python-notebook

Hämta anteckningsbok

Utforska på egen hand

Pipelinen klassificerar dokument i fem avtalstyper men extraherar fält endast consulting_agreementför . 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?.