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

Important

This feature is in Beta. Workspace admins can control access to this feature from the Previews page. See Manage Azure Databricks previews.

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

The result is a medallion-style pipeline: bronze (raw external 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 permissions to create tables in a schema and to create a pipeline.
  • Use the Preview channel.

The samples.sec.contracts dataset is available in all workspaces by default, so no additional setup is required. Because the files already live in a Unity Catalog volume, this tutorial stores them as FILE EXTERNAL references without copying their contents. 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 external 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 EXTERNAL references each file in place, without copying its contents.

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');

Python

from pyspark import pipelines as dp

@dp.table(
  name="raw_contracts",
  schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE EXTERNAL"
)
def raw_contracts():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .load("/Volumes/samples/sec/contracts/")
  )
  • Works for large files: a large PDF stays in the volume, 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.
  • 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")
  )

Note

Defining the parse step as a materialized view over the raw_contracts streaming table incrementalizes the computation. Each pipeline update runs ai_parse_document only on the files added since the last update, not on the entire table. Because ai_parse_document is the most expensive step, this avoids reparsing documents you've already processed. Incremental refresh of materialized views requires serverless compute; run the pipeline on serverless. See Spark Declarative Pipelines.

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

Tip

To improve classification accuracy, add label descriptions and an instructions option to ai_classify. See ai_classify function.

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 external FILE references, ai_parse_document and ai_classify route each document, and the consulting_agreements gold materialized view surfaces the extracted fields.

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

  • Last updated on