Skip to main content
Last updated on

Process files with UDFs

Beta

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

Use a user-defined function (UDF) to process the files referenced by a FILE column with your own code and libraries. The UDF receives each FILE value as a language-native file reference. It can read the file's bytes or open it as a local path, then return a metadata value, a derived file, or transformed output.

This page shows file-processing UDFs in Python, Scala, and SQL. For the FILE type reference, see FILE type. For general UDF authoring, see Python scalar user-defined functions (UDFs), Session-scoped Scala and Java UDFs, and Python user-defined table functions (UDTFs).

Read file metadata in a UDF​

A FILE value has metadata fields that you can read without opening the file. The following table contains the available fields:

Accessor

Description

uri

The URI of the file.

offset

An offset into the file, in bytes.

size

The size of the file, in bytes.

content_type

The MIME type of the file, when known.

checksum

A checksum used to identify the file version, as <algorithm>:<value>.

Accessor

Description

uri

The URI of the file.

offset

An offset into the file, in bytes.

size

The size of the file, in bytes.

content_type

The MIME type of the file, when known.

checksum

A checksum used to identify the file version, as <algorithm>:<value>.

Access these fields with dot notation on the FILE value, as shown in the following code:

Python
from pyspark.sql.functions import col, udf
from pyspark.sql.types import BooleanType

@udf(returnType=BooleanType())
def is_large_image(file):
  return file.content_type.startswith("image/") and file.size > 5_000_000

spark.read.table("documents").select(col("file").uri, is_large_image(col("file"))).display()

Read file contents in a UDF​

A FILE value has two methods for reading the underlying file:

  • as_local_file(): Returns a local path that you can pass to any library that accepts a file path, such as an image or media library.
  • open(): Returns a binary stream that reads only the bytes you request, instead of materializing the whole file.

Both require Databricks compute (a notebook or UDF worker) and aren't available on a Databricks Connect client. You can declare FILE as a UDF parameter or return type in Python, Scala, and SQL UDFs. For the full API, see FileType.

Extract image dimensions​

You can use a scalar UDF to return an image's dimensions as a width x height string. The UDF calls as_local_file() to get a local path, then passes that path to a standard image library (PIL in Python, ImageIO in Scala), as shown in the following code:

Python
from pyspark.sql.functions import col, udf
from pyspark.sql.types import StringType
from PIL import Image

@udf(returnType=StringType())
def image_resolution(file):
  # as_local_file() returns a pathlib.Path.
  with Image.open(file.as_local_file()) as img:
    return f"{img.width}x{img.height}"

spark.read.table("images").select(col("photo").uri, image_resolution(col("photo"))).display()

Detect a file's type from its bytes​

The following UDF reads only the first eight bytes of each file with open() and detects the file type from its magic number, without materializing the whole file:

Python
from pyspark.sql.functions import col, udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def file_signature(file):
  with file.open() as f:
    header = f.read(8)
  if header.startswith(b"%PDF"):
    return "pdf"
  if header.startswith(b"\x89PNG"):
    return "png"
  if header.startswith(b"\xff\xd8\xff"):
    return "jpeg"
  return "unknown"

spark.read.table("documents").select(col("file").uri, file_signature(col("file"))).display()

Generate multiple files with a table UDF (UDTF)​

To turn one input file into many output files, such as when splitting a video into frames, use a table UDF (UDTF). The UDTF takes a FILE as input and yields one row per output file, creating each file with FileRef.from_bytes(). Declare the file column as FILE in the UDTF's returnType schema. For general UDTF authoring, see Python user-defined table functions (UDTFs).

When a UDTF (or any UDF) writes new files with FileRef.from_bytes, your code must meet the following requirements:

  • Create the target volume before you run the UDTF. A Python worker can't create a top-level volume. Create it with CREATE VOLUME IF NOT EXISTS. Inside an existing volume, os.makedirs() can create subdirectories, but not the volume itself.
  • Pass an absolute dbfs: path. Returning a FileRef to a Delta Lake table requires a dbfs: URI, such as dbfs:/Volumes/my_catalog/my_schema/frames/frame_00000.jpg. A bare path raises DELTA_VIOLATE_CONSTRAINT_WITH_VALUES.
  • Verify writes are idempotent. Delete or skip files that already exist before writing. Because FileRef.from_bytes writes with exclusive-create flags, writing over an existing file raises FileExistsError.

Example: Extract video frames​

The following UDTF reads a video FILE, extracts each frame with the av (PyAV) library, writes it to a volume, and yields one row per frame:

Python
import io
import os
import av
from pyspark.sql.functions import udtf
from pyspark.sql.types import FileRef

@udtf(returnType="clip_id STRING, frame_index INT, frame FILE")
class ExtractFrames:
    def __init__(self):
        self.output_dir = "/Volumes/my_catalog/my_schema/frames/"
        os.makedirs(self.output_dir, exist_ok=True)

    def eval(self, video):
        clip_id = video.uri.split("/")[-1].split(".")[0]
        container = av.open(video.as_local_file())
        stream = container.streams.video[0]
        for i, frame in enumerate(container.decode(stream)):
            buffer = io.BytesIO()
            frame.to_image().save(buffer, format="JPEG")

            local_path = os.path.join(self.output_dir, f"{clip_id}_frame_{i:05d}.jpg")
            if os.path.exists(local_path):
                os.remove(local_path)

            yield (
                clip_id,
                i,
                FileRef.from_bytes(buffer.getvalue(), path=f"dbfs:{local_path}", content_type="image/jpeg"),
            )
        container.close()

spark.udtf.register("extract_frames", ExtractFrames)

Create the target table with a FILE EXTERNAL column, then call the UDTF with LATERAL to expand each video into one row per frame:

SQL
CREATE TABLE my_catalog.my_schema.drive_frames (
  clip_id STRING,
  frame_index INT,
  frame FILE EXTERNAL
);

INSERT INTO my_catalog.my_schema.drive_frames
  SELECT *
  FROM my_catalog.my_schema.drive_clips AS c
  JOIN LATERAL extract_frames(c.video) AS f;

Govern FILE columns with row filters​

Govern a FILE column with row filters based on the caller's identity or the file's metadata.

Row filter​

A row filter is a UDF that returns a BOOLEAN. Rows for which it returns false are omitted from query results.

The following row filter keeps only rows with files that reference an Excel spreadsheet, based on the file's content_type metadata:

SQL
CREATE FUNCTION excel_only(file FILE)
  RETURN file.content_type IN (
    'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
    'application/vnd.ms-excel');

ALTER TABLE documents SET ROW FILTER excel_only ON (file);

For more information on applying and managing row filters, including the Catalog Explorer steps and limitations, see Manually apply row filters and column masks.

Register a UDF in Unity Catalog​

Register a file-processing UDF in Unity Catalog to govern it with catalog permissions and reuse it across notebooks, queries, and users. Registering and running a UDF requires the following privileges:

  • To create a UDF: USAGE and CREATE on the schema, and USAGE on the catalog.
  • To run a UDF: EXECUTE on the UDF, and USAGE on the schema and catalog.

The following example registers a SQL UDF that returns a file's extension, then calls the UDF to create a new column:

SQL
CREATE FUNCTION my_catalog.my_schema.file_extension(file FILE)
  RETURNS STRING
  RETURN lower(element_at(split(file.uri, '\\.'), -1));

SELECT file.uri, my_catalog.my_schema.file_extension(file) AS extension
  FROM documents;

To register a Python or Scala UDF in Unity Catalog, see SQL and Python user-defined functions (UDFs) in Unity Catalog and Python user-defined table functions (UDTFs) in Unity Catalog.

Security: UDFs run with the owner's privileges​

UDF code runs with the privileges of the function's owner, not the function caller. The owner's privileges apply to reading the bytes of a FILE. A caller with only EXECUTE permissions on the UDF, and no direct access to the underlying volume, can still trigger reads of the referenced files.

Because a file-processing UDF is a governed access path to file contents, consider the following security and governance side effects:

  • Users can access file contents using the UDF. Grant EXECUTE permissions only to users that you intend to give indirect access to file contents.
  • Callers inherit the owner's file access. Verify that the UDF's owner has volume access no broader than what callers should have.

For more information about how Databricks determines the authorized user as execution crosses into a UDF body, see Authorized user and session user.

Next steps​

On this page