UDF付きのプロセスファイル

Important

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

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

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

UDFでファイルメタデータを読み取る

FILEの値はメタデータフィールドがあり、ファイルを開かずに読み取ることができます。 以下の表は利用可能なフィールドを示しています:

アクセサ	Description
uri	ファイルのURIです。
offset	ファイルへのオフセット、バイト単位です。
size	ファイルのサイズ (バイト単位)。
content_type	ファイルの種類(分かっている場合)です。
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()

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;

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

FILE値は基礎ファイルを読み取るために2つの方法を持ちます。

• as_local_file(): は、画像やメディアライブラリなど、ファイルパスを受け入れるライブラリに渡せるローカルパスを返します。
• open(): は、リクエストしたバイトだけを読み取るバイナリストリームを返します。ファイル全体を物質化するわけではありません。

どちらもAzure Databricksのコンピュート(ノートブックかUDFワーカー)を必要とし、Azure Databricks Connectクライアントでは利用できません。 FILEをPython、Scala、SQL UDFでUDFパラメータまたはリターン型として宣言できます。 APIの全内容については、 FileTypeを参照してください。

画像寸法の抽出

スカラー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()

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

ファイルのバイトからファイルの種類を検出します

以下の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()

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

テーブル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を返すには、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フレームあたり1行を生成します。

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行に展開します:

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列をガバメントします

呼び出し元の識別やファイルのメタデータに基づく行フィルターで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);

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

行フィルターの適用および管理、カタログエクスプローラーの手順や制限については、「 行フィルターとカラムマスクを手動で適用する」をご覧ください。

UnityカタログでUDFを登録する

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

• UDFを作成するには、スキーマ上で USAGE と CREATE 、カタログ上で USAGE します。
• UDFを実行するには、UDF EXECUTE 、スキーマとカタログを USAGE します。

以下の例は、ファイルの拡張子を返すSQL UDFをレジスタジカルし、そのUDFを呼び出して新しいカラムを作成します。

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カタログのPythonユーザー定義テーブル関数(UDTFs)を参照してください。

セキュリティ:UDFは所有者の権限で動作します

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

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

• ユーザーはUDFを使ってファイルの内容にアクセスできます。 ファイルの内容に間接アクセスを許可する意図があるユーザーにのみ EXECUTE 権限を付与してください。
• 発信者は所有者のファイルアクセス権を継承します。 UDFの所有者が発信者の範囲を超えないボリュームアクセス権を持っているか確認してください。

実行がUDFボディに移行する際にAzure Databricksがどのように認証ユーザーを決定するかの詳細については、「Authorized user」および「セッションユーザー」をご覧ください。

次のステップ

• FILE 型
• ai_parse_document 関数
• FileType
• Unity カタログの SQL および Python ユーザー定義関数 (UDF)
• 承認されたユーザーとセッション ユーザー
• 行フィルターとカラムマスクを手動で適用してください
• Pythonスカラーユーザー定義関数(UDF)
• Pythonユーザー定義テーブル関数(UDTF)
• ファイルタイプと非構造化データ