Skip to main content
Last updated on

Process files with UDFs

Beta

This feature is in Beta.

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. In Python, that reference is a FileRef object that you can import from pyspark.sql.types. The UDF 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, FileRef

@udf(returnType=BooleanType())
def is_large_image(file: FileRef) -> bool:
  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.

note

A UDF registered in Unity Catalog (CREATE FUNCTION) can read a FILE's metadata, but not its contents, and it can't create files. Use a session-scoped UDF to read a file's contents (open, as_local_file) or create one (from_bytes, from_local_file).

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 FileRef, StringType
from PIL import Image

@udf(returnType=StringType())
def image_resolution(file: FileRef) -> str:
  # 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 FileRef, StringType

@udf(returnType=StringType())
def file_signature(file: FileRef) -> str:
  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: FileRef):
        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.

Next steps​

On this page