Tutorial: Bouw een bestandsverwerkingspijplijn met het type FILE

Important

Deze functie bevindt zich in de bètaversie. Om het te gebruiken, moet een workspace-beheerder Bestandstype inschakelen op de pagina Voorvertoningen. Zie Azure Databricks previews beheren.

In deze tutorial bouw je een medallionsysteem met Lakeflow pipeline, waarmee je ongestructureerde documenten van begin tot eind verwerkt. Dit voorbeeld gebruikt de samples.sec.contracts voorbeelddataset, een verzameling door SEC ingediende juridische overeenkomsten die als PDF's in een Unity Catalog-volume zijn opgeslagen.

De pipeline neemt de PDF's als beheerde FILE referenties in met Auto Loader, parseert elk document met AI-functies, classificeert het in een overeenkomsttype en extraht gestructureerde velden voor elk type.

Voor de typereferentie, zie FILE type.

In deze tutorial, zul je:

Het resultaat is een pipeline volgens het medaillonmodel: brons (ruwe beheerde FILE referenties), zilver (geanalyseerde en geclassificeerde documenten) en goud (gewonnen velden per overeenkomsttype). Zie Wat is de medallion lakehouse-architectuur? voor meer informatie. De bronze-laag is een streaming-tabel die bestanden incrementeel opneemt, en de silver- en gold-lagen zijn gematerialiseerde weergaven die pas opnieuw berekenen wanneer hun invoer verandert.

Requirements

Als u deze zelfstudie wilt voltooien, moet u aan de volgende vereisten voldoen:

  • Wees aangemeld bij een Azure Databricks-werkruimte met Unity Catalog ingeschakeld.
  • Laat het FILE type voor je werkruimte ingeschakeld staan. Workspace-beheerders kunnen het inschakelen via de Previews-pagina. Zie Azure Databricks previews beheren.
  • Machtigingen hebben om tabellen in een schema te maken en een pipeline te maken.
  • Zorg voor een Unity Catalog-volume waar je naartoe kunt schrijven. Je definieert dit volume als de FileSpace van de bronze table, en Unity Catalog kopieert de ingevoerde bestanden erin als managed storage.
  • Gebruik het Preview-kanaal.

De samples.sec.contracts dataset is standaard beschikbaar in alle werkruimtes. Deze tutorial slaat de geïngeste PDF's op als FILE MANAGED referenties: Unity Catalog kopieert elk bestand naar het volume dat je als FileSpace van de tabel hebt aangemerkt en beheert het met de tabel, dus het verwijderen van rijen maakt de gerefereerde bestanden geschikt voor garbage collection en de tabel en de bestanden blijven synchroon. Om de pipeline aan te passen aan je eigen PDF's, wijs je het bronpad naar een volume dat je bestanden bevat. Voor andere invoeropties, zie Bestanden opnemen als het FILE-type.

Maak de bestandsverwerkingspijplijn aan

De pijplijn verwerkt documenten in drie fasen.

Stap 1. Brons: ruwe pdf's opnemen als beheerde bestandsverwijzingen

Gebruik Auto Loader om de contract-PDF's stapsgewijs van het volume te lezen. Het lezen van bestanden met format => 'file' vangt een referentie en metadata voor elk bestand zonder de bytes te materialiseren. Door de kolom als FILE MANAGED te declareren, kopieer je elk bestand naar de FileSpace van de tabel, het volume dat je met de databricks.filespace-preview tabeleigenschap instelt, zodat Unity Catalog de bestanden met de tabel beheert.

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/")
  )
  • Werkt voor grote bestanden: een grote PDF staat in de FileSpace van de tabel, terwijl de tabelrij alleen een lichtgewicht FILE-referentie opslaat (uri, size, content_type, checksum). Vergelijk dit met het BINARY type, dat de bytes inline in de rij opslaat.
  • Beheerde bestandslevenscyclus: Unity Catalog kopieert elk ingevoerd bestand naar de FileSpace van de tabel en beheert dit samen met de tabel: het verwijderen van rijen maakt de bestanden waarnaar wordt verwezen geschikt voor garbage collection, zodat de tabel en de bestanden synchroon blijven. Zie FILE MANAGED en FILE EXTERNAL voor meer informatie.
  • Incrementele verwerking: de streamingtabel voert incrementeel nieuwe bestanden in zodra ze in de bron aankomen, zonder bestaande bestanden opnieuw te verwerken. De samples.sec.contracts dataset in dit voorbeeld is statisch, maar met een live bron worden bij elke pipeline-update nieuwe bestanden opgepikt. Om ook bronwijzigingen en -verwijderingen te verspreiden, consumeer je de change feed met AUTO CDC. Zie Updates en verwijderingen toepassen bij AUTO CDC.

Stap 2. Zilver: documenten analyseren en classificeren

Geef elke FILE door aan de ai_parse_document-functie om de ruwe PDF om te zetten in een gestructureerde VARIANT met documentelementen, lay-outmetadata en tekst. Omdat ai_parse_document een FILE-kolom accepteert, leest het het document rechtstreeks uit de opslag en laadt het de bytes nooit in het clustergeheugen.

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

Het definiëren van de parsestap als een gematerialiseerde weergave over de raw_contracts streamingtabel maakt de berekening incrementeel. Elke pipeline-update draait ai_parse_document alleen op de bestanden die sinds de laatste update zijn toegevoegd, niet op de hele tabel. Omdat ai_parse_document de duurste stap is, voorkom je het opnieuw parseren van documenten die je al hebt verwerkt. Incrementele verversing van gematerialiseerde weergaven vereist serverloze compute; draai de pipeline op serverloze compute. Zie Spark Declarative Pipelines op Lakeflow.

Vervolgens geef je de geparseerde output door aan de functie ai_classify om elk document een van de vijf overeenkomsttypen toe te wijzen. Documenten met parseringsfouten worden gefilterd vóór classificatie. Dit voorbeeld zet ai_classify vast op versie 2.1, die de classificatie als een per-label object teruggeeft, dus lees het label uit de sleutel 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

Om de classificatienauwkeurigheid te verbeteren, voeg labelbeschrijvingen en een instructions-optie toe aan ai_classify. Zie ai_classify de functie.

Stap 3. Goud: velden extraheren per overeenkomsttypen

Elk overeenkomsttype heeft zijn eigen set relevante velden. Filter de geclassificeerde documenten op één type, geef de geparste inhoud door aan de ai_extract-functie met een schema van de velden die je wilt, en zet de respons vervolgens om in getypeerde kolommen. Dit voorbeeld zet ai_extract vast op versie 2.1, waarin elk geëxtraheerd veld een object is, dus lees de sleutel value.

Het volgende voorbeeld vormt de gold table voor adviesovereenkomsten:

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

Met deze statements heb je een volledig incrementele pipeline: zodra nieuwe contract-PDF's in het volume binnenkomen, voert Auto Loader ze in als beheerde FILE-referenties, ai_parse_document en ai_classify elk document routeren, waarna de consulting_agreements goudgematerialiseerde weergave de geëxtraheerde velden zichtbaar maakt.

Voorbeeldnotitieboeken

De volgende notebooks bevatten de volledige pijplijn uit deze tutorial. Deze notebooks zijn pipeline-broncode, geen uitvoerbare notebooks. Importeer het notebook voor jouw taal, en specificeer vervolgens het pad in het Source Code-veld wanneer je de pipeline configureert. Zie Pijplijnen configureren.

SQL

Bestandsverwerkingspijplijn SQL-notebook

Notebook krijgen

Python

Bestandsverwerkingspijplijn Python notebook

Notebook krijgen

Ontdek het zelf

De pijplijn classificeert documenten in vijf overeenkomsttypes, maar extraheert velden voor slechts consulting_agreement. Om deze uit te breiden, herhaal je de gold step voor elk overgebleven type, waarbij je het contract_type filter en het ai_extract schema aanpast zodat ze overeenkomen met de velden die relevant zijn voor dat type. Voorbeeld:

  • 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

Aanvullende bronnen

  • Laatst bijgewerkt om