Pular para o conteúdo principal
Última atualização em

Processar arquivos com UDFs

info

Beta

Esse recurso está em Beta.

Use uma função definida pelo usuário (UDF) para processar os arquivos referenciados por uma coluna FILE com seu próprio código e bibliotecas. O UDF recebe cada valor FILE como uma referência de arquivo nativa da linguagem. Em Python, essa referência é um objeto FileRef que você pode importar de pyspark.sql.types. O UDF pode ler os bytes do arquivo ou abri-lo como um caminho local e, em seguida, retornar um valor de metadados, um arquivo derivado ou uma saída transformada.

Esta página mostra UDFs de processamento de arquivos em Python, Scala e SQL. Para a referência do tipo FILE, consulte tipoFILE. Para a criação geral de UDFs, consulte Funções escalares definidas pelo usuário (UDFs) em Python, UDFs Scala e Java com escopo de sessão e Funções de tabela definidas pelo usuário (UDTFs) em Python.

Ler metadados de arquivo em um UDF​

Um valor FILE possui campos de metadados que você pode ler sem abrir o arquivo. A tabela a seguir contém os campos disponíveis:

Acessor

Descrição

uri

O URI do arquivo.

offset

Um offset no arquivo, em bytes.

size

O tamanho do arquivo, em bytes.

content_type

O tipo MIME do arquivo, quando conhecido.

checksum

Uma soma de verificação usada para identificar a versão do arquivo, como <algorithm>:<value>.

Acessor

Descrição

uri

O URI do arquivo.

offset

Um offset no arquivo, em bytes.

size

O tamanho do arquivo, em bytes.

content_type

O tipo MIME do arquivo, quando conhecido.

checksum

Uma soma de verificação usada para identificar a versão do arquivo, como <algorithm>:<value>.

Acesse estes campos com notação de ponto no valor FILE, conforme mostrado no código a seguir:

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

Ler o conteúdo do arquivo em um UDF​

Um valor FILE tem dois métodos para ler o arquivo subjacente:

  • as_local_file(): Retorna um caminho local que você pode passar para qualquer biblioteca que aceite um caminho de arquivo, como uma biblioteca de imagens ou de mídia.
  • open(): Retorna uma transmissão binária que lê apenas os bytes solicitados, em vez de materializar o arquivo inteiro.

Ambos exigem compute do Databricks (um notebook ou um worker de UDF) e não estão disponíveis em um cliente Databricks Connect. Você pode declarar FILE como um parâmetro de UDF ou tipo de retorno em UDFs Python, Scala e SQL. Para a API completa, consulte Tipo de Arquivo.

nota

Uma UDF registrada no Unity Catalog (CREATE FUNCTION) pode ler os metadados de um FILE, mas não o conteúdo, e não pode criar arquivos. Use uma UDF com escopo de sessão para ler o conteúdo de um arquivo (open, as_local_file) ou criar um (from_bytes, from_local_file).

Extrair dimensões da imagem​

Você pode usar uma UDF escalar para retornar as dimensões de uma imagem como uma string width x height. A UDF chama as_local_file() para obter um caminho local e, em seguida, passa esse caminho para uma biblioteca de imagens padrão (PIL em Python, ImageIO em Scala), conforme mostrado no código a seguir:

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

Detectar o tipo de um arquivo a partir de seus bytes​

O UDF a seguir lê apenas os primeiros oito bytes de cada arquivo com open() e detecta o tipo de arquivo a partir de seu número mágico, sem materializar o arquivo inteiro:

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

Gerar vários arquivos com uma UDF de tabela (UDTF)​

Para transformar um arquivo de entrada em vários arquivos de saída, como ao dividir um vídeo em quadros, use uma UDF de tabela (UDTF). A UDTF recebe um FILE como entrada e gera uma linha por arquivo de saída, criando cada arquivo com FileRef.from_bytes(). Declare a coluna de arquivo como FILE no esquema returnType da UDTF. Para a criação geral de UDTFs, consulte Funções de tabela definidas pelo usuário (UDTFs) em Python.

Quando uma UDTF (ou qualquer UDF) grava novos arquivos com FileRef.from_bytes, seu código deve atender aos seguintes requisitos:

  • Crie o volume de destino antes da execução do UDTF. Um worker Python não pode criar um volume de nível superior. Crie-o com CREATE VOLUME IF NOT EXISTS. Dentro de um volume existente, os.makedirs() pode criar subdiretórios, mas não o volume em si.
  • Passe um caminho dbfs: absoluto. Retornar um FileRef para uma tabela Delta Lake requer um URI dbfs:, como dbfs:/Volumes/my_catalog/my_schema/frames/frame_00000.jpg. Um caminho simples gera DELTA_VIOLATE_CONSTRAINT_WITH_VALUES.
  • Verifique se as gravações são idempotentes. Exclua ou ignore arquivos que já existem antes de gravar. Como FileRef.from_bytes grava com sinalizadores de criação exclusiva, gravar sobre um arquivo existente gera FileExistsError.

Exemplo: Extrair quadros de vídeo​

A UDTF a seguir lê um vídeo FILE, extrai cada quadro com a biblioteca av (PyAV), grava-o em um volume e retorna uma linha por quadro:

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)

Crie a tabela de destino com uma coluna FILE EXTERNAL, em seguida, chame a UDTF com LATERAL para expandir cada vídeo em uma linha por quadro:

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;

Controlar colunas FILE com filtros de linha​

Governe uma coluna FILE com filtros de linha com base na identidade do chamador ou nos metadados do arquivo.

Filtro de linha​

Um filtro de linha é uma UDF que retorna um BOOLEAN. As linhas para as quais retorna false são omitidas dos resultados da query.

O filtro de linha a seguir mantém apenas as linhas com arquivos que fazem referência a uma planilha do Excel, com base nos metadados content_type do arquivo:

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

Para mais informações sobre a aplicação e o gerenciamento de filtros de linha, incluindo os passos do Explorador de Catálogos e as limitações, consulte Aplicar manualmente filtros de linha e máscaras de coluna.

Registro de uma UDF no Unity Catalog​

Registre um UDF de processamento de arquivos no Unity Catalog para governá-lo com permissões de catálogo e reutilizá-lo em notebooks, queries e usuários. O registro e a execução de um UDF exigem os seguintes privilégios:

  • Para criar uma UDF: USAGE e CREATE no esquema, e USAGE no catálogo.
  • Para execução de uma UDF: EXECUTE na UDF e USAGE no esquema e no catálogo.

O exemplo a seguir registra uma UDF SQL que retorna a extensão de um arquivo e, em seguida, chama a UDF para criar uma nova coluna:

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;

Para registrar uma UDF Python ou Scala no Unity Catalog, consulte Funções definidas pelo usuário (UDFs) SQL e Python no Unity Catalog e Funções de tabela definidas pelo usuário (UDTFs) Python no Unity Catalog.

Próximos passos​

Nesta página