Important
이 기능은 베타 버전으로 제공됩니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. 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 |
파일의 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()
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 기본 파일을 읽는 두 가지 방법이 있습니다:
-
as_local_file(): 이미지 또는 미디어 라이브러리 등 파일 경로를 허용하는 라이브러리에 전달할 수 있는 로컬 경로를 반환합니다. -
open(): 전체 파일을 물질화하지 않고 요청한 바이트만 읽는 이진 스트림을 반환합니다.
두 버전 모두 Azure Databricks 컴퓨트(노트북 또는 UDF 작업자)를 필요로 하며, Azure Databricks Connect 클라이언트에서는 사용할 수 없습니다. Python, Scala, SQL UDF에서는 UDF 매개변수나 반환 타입으로 선언 FILE 할 수 있습니다. 전체 API에 대해서는 FileType을 참조하세요.
이미지 크기 추출
스칼라 UDF를 사용해 이미지의 크기를 문자열로 width x height 반환할 수 있습니다. UDF는 로컬 경로를 호출 as_local_file() 한 후, 그 경로를 표준 이미지 라이브러리(PILPython, ImageIO Scala)로 전달합니다. 다음 코드에 나와 있습니다:
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) 테이블로 여러 파일을 생성하기
예를 들어 비디오를 프레임으로 나눌 때처럼 입력 파일을 여러 개의 출력 파일로 변환하려면 테이블 UDF(UDTF)를 사용합니다. UDTF는 a 를 FILE 입력으로 받아 출력 파일 당 한 행을 생성하며, 각 파일을 로 생성합니다.FileRef.from_bytes() 파일 열을 UDTF FILE 스키마와 같이 returnType 선언하세요. 일반적인 UDTF 작성에 대해서는 Python 사용자 정의 테이블 함수(UDTFs)를 참조하세요.
UDTF(또는 다른 UDF)가 FileRef.from_bytes새 파일을 작성할 때, 코드는 다음 요구사항을 충족해야 합니다:
- UDTF를 실행하기 전에 목표 볼륨을 만드세요. Python 작업자는 최상위 볼륨을 만들 수 없습니다. 로
CREATE VOLUME IF NOT EXISTS생성하세요. 기존 볼륨os.makedirs()내부에서는 하위 디렉터리를 만들 수 있지만, 볼륨 자체는 만들 수 없습니다. - 절대
dbfs:적인 길을 지나가세요. aFileRef를 델타 호수 테이블로 반환하려면 URI가 필요dbfs:하며, 예를dbfs:/Volumes/my_catalog/my_schema/frames/frame_00000.jpg들어 . 맨길이 올라간다DELTA_VIOLATE_CONSTRAINT_WITH_VALUES. - 검증 쓰기는 놪등성입니다. 작성하기 전에 이미 존재하는 파일은 삭제하거나 건너뛰세요. 독점-생성 플래그로 쓰기 때문에
FileRef.from_bytes기존 파일 위에 쓰면 가 올라갑니다FileExistsError.
예시: 비디오 프레임 추출
다음 UDTF는 비디오를 FILE읽고, (PyAV) 라이브러리로 av 각 프레임을 추출한 뒤 볼륨에 기록하고, 프레임당 한 행을 생성합니다:
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 대상 테이블을 만든 후, UDTF LATERAL 를 호출해 각 영상을 프레임당 한 행으로 확장하세요:
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를 적용한 열을 거버넌스합니다.
행 필터
행 필터는 . 를 반환하는 BOOLEANUDF입니다. 반환 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)
}
Catalog Explorer 단계와 제한 사항을 포함한 행 필터 적용 및 관리에 대한 자세한 내용은 '수동으로 행 필터 및 열 마스크 적용'을 참조하세요.
Unity 카탈로그에서 UDF를 등록하세요
Unity 카탈로그에 파일 처리 UDF를 등록하여 카탈로그 권한으로 관리하고, 노트북, 쿼리, 사용자 간에 재사용하세요. UDF를 등록하고 실행하려면 다음과 같은 권한이 필요합니다:
- UDF를 생성하기 위해:
USAGE스키CREATE마USAGE와 카탈로그 위에서. - UDF
EXECUTE를 실행하려면 UDFUSAGE와 스키마, 카탈로그 모두에서 실행할 수 있습니다.
다음 예시는 파일의 확장자를 반환하는 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 카탈로그에서 Python 또는 Scala UDF를 등록하려면 Unity 카탈로그의 SQL 및 Python 사용자 정의 함수(UDF)와 Unity 카탈로그의 Python 사용자 정의 테이블 함수(UDTFs)를 참조하세요.
보안: UDF는 소유자의 권한으로 실행됩니다
UDF 코드는 함수 호출자의 권한이 아니라 함수 소유자의 권한으로 실행됩니다. 소유자의 권한 FILE은 . UDF에 대한 권한이 없 EXECUTE 고 기본 볼륨에 직접 접근할 수 없는 호출자는 참조된 파일의 읽기를 트리거할 수 있습니다.
파일 처리 UDF는 파일 내용에 대한 통제된 접근 경로이므로, 다음과 같은 보안 및 거버넌스 부작용을 고려할 수 있습니다:
- 사용자는 UDF를 통해 파일 내용에 접근할 수 있습니다. 파일 내용에 간접적으로 접근할 의도가 있는 사용자에게만 권한을 부여하세요
EXECUTE. - 발신자는 소유자의 파일 접근 권한을 상속받습니다. UDF 소유자가 발신자가 가져야 할 범위 이상으로 볼륨 접근 권한을 갖고 있는지 확인하세요.
실행이 UDF 바디로 넘어갈 때 Azure Databricks가 권한 있는 사용자를 어떻게 결정하는지에 대한 자세한 내용은 Authorized user and session user를 참조하세요.