File di processo con UDF

Importante

Questa funzionalità è in versione beta. Gli amministratori dell'area di lavoro possono controllare l'accesso a questa funzionalità dalla pagina Anteprime . Vedere Gestire le anteprime di Azure Databricks.

Usa una funzione definita dall'utente (UDF) per elaborare i file citati da una FILE colonna con il tuo codice e librerie personali. L'UDF riceve ogni FILE valore come riferimento a file nativo della lingua. Può leggere i byte del file o aprirlo come percorso locale, poi restituire un valore di metadati, un file derivato o un output trasformato.

Questa pagina mostra UDF per l'elaborazione dei file in Python, Scala e SQL. Per il FILE riferimento tipografico, vedi FILE tipo. Per la creazione generale di UDF, vedi funzioni definite dall'utente (UDF) scalari Python, UDF Scala e Java con ambito di sessione, e funzioni di tabella definite dall'utente (UDTF) in Python.

Leggi i metadati dei file in un UDF

Un valore ha campi FILE metadati che puoi leggere senza aprire il file. La tabella seguente contiene i campi disponibili:

Funzione Description
uri L'URI del file.
offset Un offset nel file, in byte.
size Dimensioni del file, in byte.
content_type Il tipo MIME del file, quando noto.
checksum Un checksum usato per identificare la versione del file, come <algorithm>:<value>.

Accedi a questi campi con la notazione a punto sul FILE valore, come mostrato nel seguente codice:

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

Scala

import org.apache.spark.sql.functions.{col, udf}

val isLargeImage = udf { (file: FileRef) =>
  file.contentType.startsWith("image/") && file.size > 5000000L
}

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

SQL

SELECT file.uri, file.content_type, file.size
  FROM documents
  WHERE file.content_type LIKE 'image/%'
    AND file.size > 5000000;

Leggi il contenuto del file in un UDF

Un FILE valore ha due metodi per leggere il file sottostante:

  • as_local_file(): Restituisce un percorso locale che puoi passare a qualsiasi libreria che accetti un percorso di file, come una libreria di immagini o multimediali.
  • open(): Restituisce un flusso binario che legge solo i byte che richiedi, invece di materializzare l'intero file.

Entrambi richiedono Azure Databricks compute (un notebook o un worker UDF) e non sono disponibili su un client Azure Databricks Connect. Puoi dichiarare FILE come parametro UDF o tipo di ritorno in Python, Scala e SQL UDF. Per l'API completa, vedi FileType.

Estrazione delle dimensioni dell'immagine

Puoi usare un UDF scalare per restituire le dimensioni di un'immagine come stringa width x height . L'UDF chiama as_local_file() per ottenere un percorso locale, poi passa quel percorso a una libreria immagine standard (PILin Python, ImageIO in Scala), come mostrato nel seguente codice:

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

Scala

import org.apache.spark.sql.functions.{col, udf}
import javax.imageio.ImageIO

val imageResolution = udf { (file: FileRef) =>
  // asLocalFile() returns a java.io.File.
  val image = ImageIO.read(file.asLocalFile())
  s"${image.getWidth}x${image.getHeight}"
}

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

Rileva il tipo di un file dai suoi byte

Il seguente UDF legge solo i primi otto byte di ciascun file con open() e rileva il tipo di file dal suo numero magico, senza materializzare l'intero 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()

Scala

import org.apache.spark.sql.functions.{col, udf}

val fileSignature = udf { (file: FileRef) =>
  // open() returns a java.io.InputStream.
  val stream = file.open()
  try {
    val header = new Array[Byte](8)
    val n = stream.read(header)
    if (n >= 4 && header(0) == '%' && header(1) == 'P' && header(2) == 'D' && header(3) == 'F') "pdf"
    else if (n >= 4 && header(0) == 0x89.toByte && header(1) == 'P' && header(2) == 'N' && header(3) == 'G') "png"
    else if (n >= 3 && header(0) == 0xFF.toByte && header(1) == 0xD8.toByte && header(2) == 0xFF.toByte) "jpeg"
    else "unknown"
  } finally {
    stream.close()
  }
}

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

Genera più file con una tabella UDF (UDTF)

Per trasformare un file di input in molti file di output, ad esempio quando si divide un video in fotogrammi, si utilizza un UDF a tabella (UDTF). L'UDTF prende un FILE input e produce una riga per ogni file di output, creando ogni file con FileRef.from_bytes(). Dichiara la colonna del file come FILE nello schema dell'UDTF returnType . Per la creazione generale di UDTF, vedi Python user-defined table functions (UDTF).

Quando un UDTF (o qualsiasi UDF) scrive nuovi file con FileRef.from_bytes, il tuo codice deve soddisfare i seguenti requisiti:

  • Crea il volume target prima di eseguire l'UDTF. Un worker Python non può creare un volume di livello superiore. Crealo con CREATE VOLUME IF NOT EXISTS. All'interno di un volume esistente, os.makedirs() si possono creare sottodirectory, ma non il volume stesso.
  • Passa un percorso assoluto dbfs: . Restituire a FileRef a una tabella Delta Lake richiede un dbfs: URI, come dbfs:/Volumes/my_catalog/my_schema/frames/frame_00000.jpg. Un percorso spoglio solleva DELTA_VIOLATE_CONSTRAINT_WITH_VALUES.
  • Verifica che le scritture siano idempotenti. Elimina o salta file che già esistono prima di scrivere. Poiché FileRef.from_bytes scrive con flag exclusive-create, scrivere sopra un file esistente genera FileExistsError.

Esempio: estrarre i fotogrammi video

Il seguente UDTF legge un video FILE, estrae ogni fotogramma con la av libreria (PyAV), lo scrive su un volume e produce una riga per fotogramma:

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)

Crea la tabella target con una FILE EXTERNAL colonna, poi chiama l'UDTF con LATERAL per espandere ogni video in una riga per fotogramma:

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;

Colonne FILE di governo con filtri a riga

Governa una FILE colonna con filtri a righe basati sull'identità del chiamante o sui metadati del file.

Filtro riga

Un filtro a riga è un UDF che restituisce un BOOLEANfile . Le righe per cui viene restituito false vengono omesse dai risultati delle query.

Il seguente filtro per righe mantiene solo le righe con file che fanno riferimento a un foglio di calcolo Excel, basandosi sui metadati del content_type file:

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

Python

from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType

@udf(returnType=BooleanType())
def excel_only(file):
  return file.content_type in (
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-excel")

Scala

import org.apache.spark.sql.functions.udf

val excelOnly = udf { (file: FileRef) =>
  Set(
    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "application/vnd.ms-excel").contains(file.contentType)
}

Per maggiori informazioni sull'applicazione e gestione dei filtri per righe, inclusi i passaggi e le limitazioni di Catalog Explor, vedi Applicare manualmente filtri a righe e maschere a colonna.

Registra un UDF nel Catalogo Unity

Registra un UDF per l'elaborazione file nel Catalogo Unity per governarlo con permessi di catalogo e riutilizzarlo tra notebook, query e utenti. La registrazione e l'esecuzione di un UDF richiedono i seguenti privilegi:

  • Per creare un UDF: USAGE e CREATE sullo schema, e USAGE sul catalogo.
  • Per eseguire un UDF: EXECUTE sull'UDF, e USAGE sullo schema e il catalogo.

Il seguente esempio registra un UDF SQL che restituisce l'estensione di un file, quindi chiama il UDF per creare una nuova colonna:

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;

Per registrare un UDF Python o Scala nel Catalogo Unity, vedi funzioni definite dall'utente (UDF) SQL e Python nel Catalogo Unity e funzioni di tabella definite dall'utente (UDTF) in Python nel Catalogo Unity.

Sicurezza: gli UDF funzionano con i privilegi del proprietario

Il codice UDF viene eseguito con i privilegi del proprietario della funzione, non del chiamante della funzione. I privilegi del proprietario si applicano alla lettura dei byte di un FILE. Un chiamante con solo EXECUTE permessi sull'UDF e senza accesso diretto al volume sottostante può comunque attivare la lettura dei file riferiti.

Poiché un UDF per l'elaborazione dei file è un percorso di accesso governato ai contenuti dei file, considera i seguenti effetti collaterali di sicurezza e governance:

  • Gli utenti possono accedere ai contenuti dei file utilizzando l'UDF. Concedi EXECUTE permessi solo agli utenti che intendi dare accesso indiretto ai contenuti dei file.
  • I chiamanti ereditano l'accesso ai file del proprietario. Verifica che il proprietario dell'UDF abbia un accesso al volume non più ampio di quello che dovrebbero avere i chiamanti.

Per maggiori informazioni su come Azure Databricks determina l'utente autorizzato quando l'esecuzione passa a un corpo UDF, vedi Utente autorizzato e utente di sessione.

Passaggi successivi

  • Last updated on