Tutorial: Build a file-processing pipeline with the FILE type

Learn how to build a medallion pipeline with Lakeflow pipeline that processes unstructured documents end to end. This example uses the samples.sec.contracts sample dataset, a collection of SEC-filed legal agreements stored as PDFs in a Unity Catalog volume.

The pipeline ingests the PDFs as managed FILE references with Auto Loader, parses each document with AI functions, classifies it into an agreement type, and extracts structured fields for each type.

For the type reference, see FILE type.

In this tutorial, you will:

• Incrementally ingest contract PDFs from a volume as managed FILE references with Auto Loader.
• Parse each document with ai_parse_document function and classify it with ai_classify function.
• Extract structured fields for each agreement type with ai_extract function.

The result is a medallion-style pipeline: bronze (raw managed FILE references), silver (parsed and classified documents), and gold (extracted fields per agreement type). See What is the medallion lakehouse architecture? for more information. The bronze layer is a streaming table that incrementally ingests files, and the silver and gold layers are materialized views that recompute only when their inputs change.

Requirements

To complete this tutorial, you must meet the following requirements:

• Be logged in to a Azure Databricks workspace with Unity Catalog enabled.
• Have the FILE type enabled for your workspace. Workspace admins can enable it from the Previews page. See Manage Azure Databricks previews.
• Have permissions to create tables in a schema and to create a pipeline.
• Have a Unity Catalog volume that you can write to. You declare this volume as the bronze table's FileSpace, and Unity Catalog copies the ingested files into it as managed storage.
• Use the Preview channel.

The samples.sec.contracts dataset is available in all workspaces by default. This tutorial stores the ingested PDFs as FILE MANAGED references: Unity Catalog copies each file into the volume you declare as the table's FileSpace and manages it with the table, so deleting rows makes the referenced files eligible for garbage collection and the table and its files stay in sync. To adapt the pipeline to your own PDFs, point the source path at a volume that contains your files. For other ingestion options, see Ingest files as the FILE type.

Create the file-processing pipeline

The pipeline processes documents in three stages.

Step 1. Bronze: ingest raw PDFs as managed FILE references

Use Auto Loader to incrementally read the contract PDFs from the volume. Reading files with format => 'file' captures a reference and metadata for each file without materializing its bytes. Declaring the column as FILE MANAGED copies each file into the table's FileSpace, the volume you set with the databricks.filespace-preview table property, so Unity Catalog manages the files with the table.

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

• Works for large files: a large PDF lives in the table's FileSpace, while the table row stores only a lightweight FILE reference (uri, size, content_type, checksum). Compare this with the BINARY type, which inlines the bytes in the row.
• Managed file lifecycle: Unity Catalog copies each ingested file into the table's FileSpace and manages it with the table: deleting rows makes the referenced files eligible for garbage collection, so the table and its files stay in sync. For details, see FILE MANAGED and FILE EXTERNAL.
• Incremental processing: the streaming table incrementally ingests new files as they arrive in the source, without reprocessing existing ones. The samples.sec.contracts dataset in this example is static, but with a live source, new files are picked up on each pipeline update. To also propagate source changes and deletions, ingest the change feed with AUTO CDC. See Apply updates and deletions with AUTO CDC.

Step 2. Silver: parse and classify documents

Pass each FILE to ai_parse_document function to convert the raw PDF into a structured VARIANT containing document elements, layout metadata, and text. Because ai_parse_document accepts a FILE column, it reads the document directly from storage and never loads the bytes into cluster memory.

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

Next, pass the parsed output to ai_classify function to assign each document one of five agreement types. Documents with parsing errors are filtered out before classification. This example pins ai_classify to version 2.1, which returns the classification as a per-label object, so read the label from the value key.

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

Step 3. Gold: extract fields per agreement type

Each agreement type has its own set of relevant fields. Filter the classified documents to one type, pass the parsed content to ai_extract function with a schema of the fields you want, then flatten the response into typed columns. This example pins ai_extract to version 2.1, in which each extracted field is an object, so read its value key.

The following example builds the gold table for consulting agreements:

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

With these statements, you have a fully incremental pipeline: as new contract PDFs arrive in the volume, Auto Loader ingests them as managed FILE references, ai_parse_document and ai_classify route each document, and the consulting_agreements gold materialized view surfaces the extracted fields.

Example notebooks

The following notebooks contain the complete pipeline from this tutorial. These notebooks are pipeline source code, not runnable notebooks. Import the notebook for your language, then specify its path in the Source code field when you configure the pipeline. See Configure pipelines.

SQL

File-processing pipeline SQL notebook

Get notebook

Python

File-processing pipeline Python notebook

Get notebook

Explore on your own

The pipeline classifies documents into five agreement types but extracts fields for only consulting_agreement. To extend it, repeat the gold step for each remaining type, changing the contract_type filter and the ai_extract schema to match the fields relevant to that type. For example:

• 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

Additional resources

• FILE type
• Ingest files as the FILE type
• FILE functions quickstart
• Learn more about Auto Loader. See What is Auto Loader?.