メインコンテンツまでスキップ
に最終更新

UDF を使用したファイルの処理

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

ユーザー定義関数（UDF）を使用して、FILE カラムによって参照されるファイルを独自のコードとライブラリで処理します。UDF は、各 FILE 値を言語ネイティブのファイル参照として受け取ります。ファイルのバイトを読み取るか、ローカルパスとして開いてから、メタデータ値、派生ファイル、または変換された出力を返すことができます。

このページでは、Python、Scala、SQL でのファイル処理 UDF を紹介します。FILE タイプのリファレンスについては、FILE タイプを参照してください。一般的な UDF の作成については、Python スカラーユーザー定義関数 (UDF)、セッションスコープの Scala および Java UDF、および Python ユーザー定義テーブル関数 (UDTF) を参照してください。

UDF でのファイル メタデータの読み取り​

FILE 値には、ファイルを開かずに読み取ることができるメタデータ フィールドがあります。次の表は、利用可能なフィールドをまとめたものです：

アクセサー

説明

uri

ファイルの URI。

offset

ファイル内のオフセット（バイト単位）。

size

ファイルのサイズ（バイト単位）。

content_type

既知の場合の、ファイルの MIME タイプ。

checksum

ファイルバージョンを識別するために使用されるチェックサム（<algorithm>:<value> など）。

アクセサー

説明

uri

ファイルの URI。

offset

ファイル内のオフセット（バイト単位）。

size

ファイルのサイズ（バイト単位）。

content_type

既知の場合の、ファイルの MIME タイプ。

checksum

ファイルバージョンを識別するために使用されるチェックサム（<algorithm>:<value> など）。

次のコードに示すように、FILE 値に対してドット表記を使用してこれらのフィールドにアクセスします。

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

UDF でファイルの内容を読み取る​

FILE 値には、基盤となるファイルを読み取るための 2 つのメソッドがあります。

  • as_local_file(): 画像ライブラリやメディアライブラリなど、ファイルパスを受け入れる任意のライブラリに渡すことができるローカルパスを返します。
  • open()：ファイル全体をマテリアライズするのではなく、要求したバイトのみを読み取るバイナリストリームを返します。

どちらも Databricks コンピュート（ノートブックまたは UDF ワーカー）を必要とし、Databricks Connect クライアントでは使用できません。FILE は、Python、Scala、および SQL UDF において、UDF パラメーターまたは戻り値の型として宣言できます。完全な API については、ファイルタイプを参照してください。

画像ディメンションの抽出​

スカラー UDF を使用して、画像の寸法を width x height 文字列として返すことができます。UDF は as_local_file() を呼び出してローカルパスを取得し、次のコードに示すように、そのパスを標準の画像ライブラリ (Python では PIL、Scala では ImageIO) に渡します。

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

バイトからファイルのタイプを検出する​

以下の UDF は、ファイル全体を具体化することなく、open() を使用して各ファイルの最初の 8 バイトのみを読み取り、マジックナンバーからファイル形式を検出します。

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

テーブルUDF（UDTF）を使用して複数のファイルを生成する​

ビデオをフレームに分割する場合のように、1 つの入力ファイルを複数の出力ファイルに変換するには、テーブル UDF (UDTF) を使用します。UDTF は FILE を入力として受け取り、出力ファイルごとに 1 行を生成し、FileRef.from_bytes() を使用して各ファイルを作成します。UDTF の returnType スキーマで、ファイルカラムを FILE として宣言します。一般的な UDTF の作成については、「Python ユーザー定義テーブル関数 (UDTF)」を参照してください。

UDTF (または任意の UDF) が FileRef.from_bytes を使用して新しいファイルを書き込む場合、コードは次の要件を満たしている必要があります。

  • UDTF をランする前に、ターゲットボリュームを作成してください。 Python ワーカーは、トップレベルのボリュームを作成できません。CREATE VOLUME IF NOT EXISTS を使用して作成します。既存のボリューム内では、os.makedirs() はサブディレクトリを作成できますが、ボリューム自体を作成することはできません。
  • 絶対dbfs:パスを渡します。 FileRefをDelta Lakeテーブルに返すには、dbfs:/Volumes/my_catalog/my_schema/frames/frame_00000.jpgのようなdbfs: URIが必要です。ベアパスはDELTA_VIOLATE_CONSTRAINT_WITH_VALUESを発生させます。
  • 書き込みがべき等であることを確認します。 書き込み前に、既に存在するファイルを削除またはスキップします。FileRef.from_bytes は排他的作成フラグを使用して書き込むため、既存のファイルに上書きすると FileExistsError が発生します。

例：ビデオフレームの抽出​

次の UDTF は、動画 FILE を読み取り、av (PyAV) ライブラリを使用して各フレームを抽出し、それをボリュームに書き込み、フレームごとに 1 行を生成します：

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)

FILE EXTERNAL 列を持つターゲットテーブルを作成し、LATERAL を指定して UDTF を呼び出し、各動画をフレームごとに 1 行に展開します：

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;

行フィルターによる FILE 列の管理​

呼び出し元のIDまたはファイルのメタデータに基づいて、行フィルターを使用して FILE 列を管理します。

行フィルター​

行フィルターは、BOOLEAN を返す UDF です。false を返す行は、クエリー結果から除外されます。

次の行フィルターは、ファイルの content_type メタデータに基づいて、Excel スプレッドシートを参照するファイルを含む行のみを保持します。

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

Catalog Explorerのステップや制限事項など、行フィルターの適用と管理の情報については、「行フィルターと列マスクを手動で適用する」を参照してください。

Unity Catalog に UDF を登録する​

Unity Catalog にファイル処理 UDF を登録して、カタログ権限で管理し、ノートブック、クエリー、ユーザー間で再利用できるようにします。UDF の登録と実行には、以下の権限が必要です。

  • UDF を作成するには：スキーマに対する USAGE および CREATE、そしてカタログに対する USAGE が必要です。
  • UDF をランするには、UDF に対して EXECUTE を実行し、スキーマとカタログに対して USAGE を実行します。 [[ ## completed ##]]

次の例では、ファイルの拡張子を返す SQL UDF を登録し、その UDF を呼び出して新しい列を作成します。

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;

Unity CatalogにPythonまたはScalaのUDFを登録するには、Unity CatalogのSQLおよびPythonユーザー定義関数 (UDF)およびUnity CatalogのPythonユーザー定義テーブル関数 (UDTF)を参照してください。

セキュリティ：UDFは所有者の権限でランします​

UDF コードは、関数を呼び出したユーザーではなく、関数の 所有者 の権限でランします。所有者の権限は、FILE のバイトの読み取りに適用されます。UDF に対する EXECUTE 権限のみを持ち、基盤となるボリュームへの直接アクセス権を持たない呼び出し元であっても、参照されているファイルの読み取りを Trigger できます。

ファイル処理 UDF はファイル・コンテンツへの管理されたアクセス・パスであるため、以下のセキュリティおよびガバナンスの副作用を考慮してください。

  • ユーザーは UDF を使用してファイルの内容にアクセスできます。 ファイルの内容への間接的なアクセス権を付与するつもりのユーザーに対してのみ、EXECUTE 権限を付与してください。
  • 呼び出し元は、所有者のファイルアクセス権を継承します。 UDF の所有者が持つボリュームアクセス権が、呼び出し元が持つべき範囲を超えていないことを確認してください。

実行が UDF 本文に移行する際に Databricks がどのように認可ユーザーを決定するかについての情報は、認可ユーザーとセッションユーザーを参照してください。

次のステップ​

このページの見出し