Ingest files as 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.

The FILE type stores and queries references to unstructured files (documents, images, and audio) in tables. This page shows how to discover files, ingest them as FILE references, and incrementally ingest new files as they arrive.

For the reference on the FILE type, see FILE type. For an overview of approaches for ingesting unstructured data, see FILE type and unstructured data.

Note

FILE columns don't have a defined ordering. You can't use a FILE column as a partitioning column, a clustering column, or a Z-order key. For more information, see Limits.

Storage modes

A FILE reference can be stored in one of two modes:

FILE EXTERNAL references files that already exist in a Unity Catalog volume. Databricks doesn't support storing FILE EXTERNAL references for files stored outside volumes.
FILE MANAGED stores copies of files in Unity Catalog-managed storage. Files from sources outside of volumes, such as SharePoint, Google Drive, or SFTP, must be ingested and stored as FILE MANAGED.

Use `list_files` to discover files

Use the list_files table-valued function table-valued function to discover the files available at a path. It returns one row per file with its path, size, modification_time, and a FILE reference:

SELECT * FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

To discover files in a source that requires a Unity Catalog connection, such as SharePoint, Google Drive, or SFTP, add the connection parameter:

SELECT * FROM list_files('https://example.sharepoint.com/sites/my-site/', connection => 'my_sharepoint_connection');

list_files discovers files recursively by default. To learn more, see list_files table-valued function.

Ingest files as FILE references

Select an ingestion approach based on where you store your files. To reference files already in a Unity Catalog volume, use FILE EXTERNAL. To ingest files from an external source, copy them into managed storage as FILE MANAGED.

Ingest volume files as FILE EXTERNAL

To ingest files that already exist in a Unity Catalog volume, use a CREATE TABLE AS SELECT (CTAS) statement with list_files. This creates a table with a FILE EXTERNAL column that references each file in place, without copying its contents. The following example creates a documents table with the file name, metadata, and a FILE reference for each file:

CREATE TABLE documents AS
  SELECT _metadata.file_name, *
  FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

Ingest external source files as FILE MANAGED

To generate FILE references for files in a source such as SharePoint, Google Drive, or SFTP, ingest the files first and store them as FILE MANAGED. FILE EXTERNAL isn't supported for files stored outside volumes.

The following example ingests files from SharePoint into a FILE MANAGED table:

SQL

CREATE TABLE managed_documents (
  file_name STRING,
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents
  SELECT _metadata.file_name, *
  FROM read_files(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file');

Python

(spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append())

Scala

spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append()

Use pipelines to incrementally ingest new files

To ingest new files as they arrive, use a streaming table in a Lakeflow pipeline that reads the source with STREAM read_files(..., format => 'file'). Each pipeline update processes only the files added after the last update. See read_files and Spark Declarative Pipelines.

To incrementally stream files from a source such as Google Drive:

Set the pipeline's channel to PREVIEW. Ingesting FILE references in a pipeline requires the PREVIEW channel.

Define a streaming table that reads the source with STREAM read_files(..., format => 'file'), as in the following code:

SQL

CREATE STREAMING TABLE streaming_documents (
  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(
    'https://drive.google.com/drive/folders/my-folder-id',
    connection => 'my_gdrive_connection',
    format => 'file');

Python

from pyspark import pipelines as dp

@dp.table(
  name="streaming_documents",
  schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)
def streaming_documents():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .option("databricks.connection", "my_gdrive_connection")
      .load("https://drive.google.com/drive/folders/my-folder-id")
  )

Apply updates and deletions with AUTO CDC

A streaming ingest adds new files but doesn't capture updates or deletions from the source. To apply those changes, read the source change feed with AUTO CDC.

Warning

Databricks recommends that you land the change data in a managed table first, as in the following example, then apply AUTO CDC to that table. Applying AUTO CDC directly to STREAM read_files(..., readChangeFeed => true) re-reads the source change feed for each downstream flow, which might increase processing costs.

Ingest the change feed in two steps. The following example ingests the change feed from SharePoint, then applies it to a target streaming table as SCD type 1:

Write the change data into a streaming table with managed files, as in the following code. Set readChangeFeed => true on read_files to return the change feed, which includes the _file_id, _sequence, and _is_deleted metadata columns.

SQL

CREATE OR REFRESH STREAMING TABLE documents_changes (
  _file_id STRING,
  _sequence BIGINT,
  _is_deleted BOOLEAN,
  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(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file',
    readChangeFeed => true);

Python

from pyspark import pipelines as dp

@dp.table(
  name="documents_changes",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)
def documents_changes():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .option("databricks.connection", "my_sharepoint_connection")
      .option("cloudFiles.readChangeFeed", "true")
      .load("https://example.sharepoint.com/sites/my-site/")
  )

Use AUTO CDC to apply the changes from that table to a target streaming table, as in the following code. Use _file_id as the key, _sequence as the sequence column, and _is_deleted to identify deletions.

SQL

CREATE OR REFRESH STREAMING TABLE documents
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

CREATE FLOW documents_cdc AS AUTO CDC INTO
  documents
FROM STREAM documents_changes
  KEYS (_file_id)
  APPLY AS DELETE WHEN _is_deleted = true
  SEQUENCE BY _sequence
  COLUMNS * EXCEPT (_is_deleted, _sequence)
  STORED AS SCD TYPE 1;

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

dp.create_streaming_table(
  name="documents",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)

dp.create_auto_cdc_flow(
  target = "documents",
  source = "documents_changes",
  keys = ["_file_id"],
  sequence_by = col("_sequence"),
  apply_as_deletes = expr("_is_deleted = true"),
  except_column_list = ["_is_deleted", "_sequence"],
  stored_as_scd_type = 1
)

Convert inline binary data to FILE references

If a table already stores file contents as inline binary data, use create_file function to write that data to storage and produce a FILE reference.

The following examples use a user generated table, raw_documents, with a name column and a content column that holds the binary data.

Write binary data to a volume as FILE EXTERNAL

To write the files to a Unity Catalog volume as external files, pass a destination_path to create_file, as in the following code:

SQL

CREATE TABLE documents (name STRING, file FILE EXTERNAL) USING DELTA;

INSERT INTO documents (name, file)
  SELECT
    name,
    create_file(
      content => content,
      destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name
    )
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append()

Write binary data to managed storage as FILE MANAGED

To store the files as managed files instead, call create_file with just the binary content. When you omit destination_path, Unity Catalog uploads the content to the managed storage location:

SQL

CREATE TABLE managed_documents (name STRING, file FILE MANAGED) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents (name, file)
  SELECT name, create_file(content => content)
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append()

Next steps

FILE type
FILE type and unstructured data
Tutorial: Build a file-processing pipeline with the FILE type
Learn more about Auto Loader. See What is Auto Loader?.

Feedback

Was this page helpful?

Last updated on 2026-08-03

Ingest files as the FILE type

Storage modes

Use list_files to discover files

Ingest files as FILE references

Ingest volume files as FILE EXTERNAL

Ingest external source files as FILE MANAGED

SQL

Python

Scala

Use pipelines to incrementally ingest new files

SQL

Python

Apply updates and deletions with AUTO CDC

SQL

Python

SQL

Python

Convert inline binary data to FILE references

Write binary data to a volume as FILE EXTERNAL

SQL

Python

Scala

Write binary data to managed storage as FILE MANAGED

SQL

Python

Scala

Next steps

Feedback

Additional resources

Use `list_files` to discover files