Python 스칼라 UDF는 Azure Databricks에서 SQL 쿼리 내에서 커스텀 Python 로직을 실행할 수 있게 해줍니다. 이 페이지는 등록 및 호출, 서비스 자격 증명 및 비밀 사용, 그리고 Spark SQL에서 서브식의 평가 순서 주의사항을 다루는 방법을 보여줍니다.
요구 사항
Databricks Runtime 12.2 LTS 이하에서는 표준 액세스 모드를 사용하는 Unity 카탈로그 컴퓨팅에서 Python UDF 및 Pandas UDF가 지원되지 않습니다.
스칼라 Python UDF 및 Pandas UDF는 모든 액세스 모드에 대해 Databricks Runtime 13.3 LTS 이상에서 지원됩니다.
Unity 카탈로그 사용 클러스터에서 Python UDF에 대한 ARM 인스턴스를 지원하려면 Databricks Runtime 15.2 이상이 필요합니다.
Databricks Runtime 14.0 이하에서는 표준 액세스 모드를 사용하는 Unity 카탈로그 클러스터에서 Python UDF 및 Pandas UDF가 지원되지 않습니다. Scalar Python UDF 및 Pandas UDF는 Databricks Runtime 14.1 이상의 모든 액세스 모드에서 지원됩니다.
Databricks Runtime 14.1 이상에서는 SQL 구문을 사용하여 Unity 카탈로그에 스칼라 Python UDF를 등록할 수 있습니다. Unity 카탈로그에서 SQL 및 UDF(사용자 정의 함수) Python 참조하세요.
UDF로 함수 등록
def squared(s):
return s * s
spark.udf.register("squaredWithPython", squared)
필요에 따라 UDF의 반환 형식을 설정할 수 있습니다. 기본 반환 형식은 StringType입니다.
from pyspark.sql.types import LongType
def squared_typed(s):
return s * s
spark.udf.register("squaredWithPython", squared_typed, LongType())
Spark SQL에서 UDF 호출
spark.range(1, 20).createOrReplaceTempView("test")
%sql select id, squaredWithPython(id) as id_squared from test
DataFrames에서 UDF 사용
from pyspark.sql.functions import udf
from pyspark.sql.types import LongType
squared_udf = udf(squared, LongType())
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))
또는 주석 구문을 사용하여 동일한 UDF를 선언할 수 있습니다.
from pyspark.sql.functions import udf
@udf("long")
def squared_udf(s):
return s * s
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))
UDF를 사용하는 변형
PySpark 변형의 형식은 VariantType이며, 그 값은 VariantVal 형식입니다. 변형에 대한 자세한 내용은 쿼리 변형 데이터를 참조하세요.
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import VariantType, VariantVal
# Return Variant
@udf(returnType = VariantType())
def toVariant(jsonString):
return VariantVal.parseJson(jsonString)
spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toVariant(col("json"))).display()
+---------------+
|toVariant(json)|
+---------------+
| {"a":1}|
+---------------+
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import StructField, StructType, VariantType, VariantVal
# Return Struct<Variant>
@udf(returnType = StructType([StructField("v", VariantType(), True)]))
def toStructVariant(jsonString):
return {"v": VariantVal.parseJson(jsonString)}
spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toStructVariant(col("json"))).display()
+---------------------+
|toStructVariant(json)|
+---------------------+
| {"v":{"a":1}}|
+---------------------+
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import ArrayType, VariantType, VariantVal
# Return Array<Variant>
@udf(returnType = ArrayType(VariantType()))
def toArrayVariant(jsonString):
return [VariantVal.parseJson(jsonString)]
spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toArrayVariant(col("json"))).display()
+--------------------+
|toArrayVariant(json)|
+--------------------+
| [{"a":1}]|
+--------------------+
from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import MapType, StringType, VariantType, VariantVal
# Return Map<String, Variant>
@udf(returnType = MapType(StringType(), VariantType(), True))
def toMapVariant(jsonString):
return {"v1": VariantVal.parseJson(jsonString), "v2": VariantVal.parseJson("[" + jsonString + "]")}
spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toMapVariant(col("json"))).display()
+-----------------------------+
| toMapVariant(json)|
+-----------------------------+
|{"v2":[{"a":1}],"v1":{"a":1}}|
+-----------------------------+
UDF가 포함된 파일
Important
이 기능은 베타 버전으로 제공됩니다.
파일의 PySpark 타입은 FileType입니다. UDF에서 매개변수나 반환 타입으로 사용하며, 최상위 타입이나 중첩 형태로 사용할 수 있습니다. 타입, 중첩 규칙, FileRef API에 대해서는 FileType을 참조하세요.
UDF에서 파일의 내용을 읽으려면, 열어볼 수 있는 로컬 경로를 호출 file.as_local_file() 하거나, file.open() 바이트를 스트림으로 읽으세요. Python, Scala, SQL의 이미지 처리, 파일 유형 감지, 비디오 프레임 추출 등의 예시는 UDF가 포함된 프로세스 파일을 참조하세요. 타입 참조에 대해서는 타입을 참조하세요FILE.
Unity Catalog(CREATE FUNCTION)에 등록된 UDF는 FILE의 메타데이터는 읽을 수 있지만 그 내용은 읽을 수 없으며, 파일을 생성할 수도 없습니다. 세션 범위 UDF를 사용하여 파일의 내용을 읽거나(open, as_local_file) 파일을 생성하세요(from_bytes, from_local_file).
UDF 본체에서 각 FILE 값은 객체입니다 FileRef :
from pyspark.sql.functions import col, udf
from pyspark.sql.types import FileRef, StringType
@udf(returnType=StringType())
def file_content_type(file: FileRef) -> str:
return file.content_type
df = spark.table("documents")
display(df.select(col("file").uri, file_content_type(col("file"))))
평가 순서 및 Null 검사
Spark SQL(SQL과 DataFrame 및 데이터 세트 API 포함)은 하위 식의 평가 순서를 보장하지 않습니다. 특히 연산자 또는 함수 입력이 반드시 왼쪽에서 오른쪽 또는 다른 고정 순서로 평가되는 것은 아닙니다. 예를 들어, 논리적 AND 및 OR 표현식에는 왼쪽에서 오른쪽으로의 "쇼트 서킷" 의미 체계가 없습니다.
따라서 쿼리 최적화 및 계획 중에 식과 절의 순서가 변경될 수 있으므로 부울 식의 부작용 또는 평가 순서와 WHERE 및 HAVING 절의 순서에 의존하는 것은 위험합니다. 특히 UDF가 Null 검사를 위해 SQL의 단락 의미 체계를 사용하는 경우 UDF를 호출하기 전에 null 검사가 수행된다는 보장은 없습니다. 예를 들면 다음과 같습니다.
spark.udf.register("strlen", lambda s: len(s), "int")
spark.sql("select s from test1 where s is not null and strlen(s) > 1") # no guarantee
여기서 WHERE 절은 Null을 필터링한 후 strlen UDF가 호출되도록 보장하지 않습니다.
적절한 Null 검사를 수행하려면 다음 중 하나를 수행하는 것이 좋습니다.
- UDF 자체가 Null을 인식하도록 설정하고 UDF 내에서 Null 검사를 수행합니다.
-
IF또는CASE WHEN식을 사용하여 Null 검사를 수행하고 조건부 분기에서 UDF를 호출합니다.
spark.udf.register("strlen_nullsafe", lambda s: len(s) if not s is None else -1, "int")
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") # ok
spark.sql("select s from test1 where if(s is not null, strlen(s), null) > 1") # ok
Unity 카탈로그 비밀 접근
세션 스코비가 적용된 Python UDF에서 Unity 카탈로그 비밀에 접근하려면 '세션 범위가 적용된 Python UDF에서 비밀 사용하기'를 참조하세요. 스칼라 또는 배치 Unity 카탈로그 Python UDF에서 선언된 비밀에 접근하려면 'Python UDF에서 비밀 사용하기'를 참조하세요.
Python UDF에서의 서비스 자격 증명
세션 범위 스칼라 Python UDF와 스칼라 Unity 카탈로그 Python UDF는 Unity 카탈로그 서비스 자격 증명을 사용하여 외부 클라우드 서비스에 안전하게 접근할 수 있습니다. 클라우드 기반 토큰화, 암호화 또는 비밀 관리와 같은 작업을 데이터 변환에 직접 통합하는 데 유용합니다.
요구 사항은 UDF 유형과 계산량에 따라 다릅니다. Python UDF에서 서비스 증명 사용하기를 참조하세요.
서비스 자격 증명을 만들려면 서비스 자격 증명 만들기를 참조하세요.
세션 범위 스칼라 Python UDF에서 서비스 자격 증명을 사용하세요
서비스 자격 증명에 액세스하려면 UDF 논리의 databricks.service_credentials.getServiceCredentialsProvider() 유틸리티를 사용하여 적절한 자격 증명으로 클라우드 SDK를 초기화합니다. 모든 코드는 UDF 본문에 캡슐화되어야 합니다.
@udf
def use_service_credential():
from azure.mgmt.web import WebSiteManagementClient
# Assuming there is a service credential named 'testcred' set up in Unity Catalog
web_client = WebSiteManagementClient(subscription_id, credential = getServiceCredentialsProvider('testcred'))
# Use web_client to perform operations
서비스 자격 증명 권한
세션 범위의 UDF는 호출자의 권한을 사용합니다. 필요한 권한은 'Python UDF에서 서비스 자격 증명 사용(Use a service credential)'을 참조하세요.
세션 범위의 UDF를 위한 컴퓨트 수준 기본 자격 증명
스칼라 Python UDF에서 사용할 때, Databricks는 컴퓨트 환경 변수의 기본 서비스 자격 증명을 자동으로 사용합니다. 이 동작을 사용하면 UDF 코드에서 자격 증명 별칭을 명시적으로 관리하지 않고도 외부 서비스를 안전하게 참조할 수 있습니다. 컴퓨팅 리소스에 대한 기본 서비스 자격 증명 지정을 참조하세요.
기본 자격 증명 지원은 표준 및 전용 액세스 모드 클러스터에서만 사용할 수 있습니다. SQL 웨어하우스에서는 제공되지 않습니다.
azure-identity 패키지를 설치하여 DefaultAzureCredential 제공자를 사용하십시오. 패키지를 설치하려면 Notebook 범위 내 Python 라이브러리 또는 컴퓨터 범위 내 라이브러리를 참조하세요.
@udf
def use_service_credential():
from azure.identity import DefaultAzureCredential
from azure.mgmt.web import WebSiteManagementClient
# DefaultAzureCredential is automatically using the default service credential for the compute
web_client_default = WebSiteManagementClient(DefaultAzureCredential(), subscription_id)
# Use web_client to perform operations
Unity Catalog Python UDF 스칼라 기반 서비스 자격 증명을 사용하세요
UDF 정의의 절에서 CREDENTIALS 서비스 자격 증명을 명시하세요. 한 가지 인증 마크 DEFAULT 를 표시하면 패치된 클라우드 SDK가 자동으로 사용하게 됩니다. 클래식 컴퓨트에서는 이 기능이 Databricks Runtime 18.1 이상이어야 합니다. 서버리스 컴퓨트와 프로 및 서버리스 SQL 웨어하우스에서는 UDF의 environment_version를 명시적으로 6 이상으로 설정하세요. 전체 컴퓨팅, 네트워킹 및 권한 요구 사항은 Python UDF에서 서비스 자격 증명 사용을 참조하세요.
작업 실행 컨텍스트 가져오기
TaskContext PySpark API를 사용하여 사용자 정보, 클러스터 태그, 스파크 작업 ID 등과 같은 컨텍스트 정보를 얻으세요. UDF에서 작업 컨텍스트 가져오기를 참조하십시오.
제한 사항
PySpark UDF에는 다음과 같은 제한 사항이 적용됩니다.
파일 액세스 제한 사항: Databricks Runtime 14.2 이하에서 공유 클러스터의 PySpark UDF는 Git 폴더, 작업 영역 파일 또는 Unity 카탈로그 볼륨에 액세스할 수 없습니다.
브로드캐스트 변수: 표준 액세스 모드 클러스터 및 서버리스 컴퓨팅의 PySpark UDF는 브로드캐스트 변수를 지원하지 않습니다.
- 서버리스의 메모리 제한: 서버리스 컴퓨팅의 PySpark UDF의 메모리 제한은 PySpark UDF당 1GB입니다. 이 제한을 초과하면 UDF_PYSPARK_USER_CODE_ERROR.MEMORY_LIMIT_SERVERLESS 형식의 오류가 발생합니다.
- 표준 액세스 모드의 메모리 제한: 표준 액세스 모드의 PySpark UDF에는 선택한 인스턴스 유형의 사용 가능한 메모리에 따라 메모리 제한이 있습니다. 사용 가능한 메모리를 초과하면 형식 UDF_PYSPARK_USER_CODE_ERROR 오류가 발생합니다 . MEMORY_LIMIT.
- 서버리스 SQL 웨어하우스의 네트워크 액세스: 기본적으로 서버리스 SQL 웨어하우스의 Python UDF는 아웃바운드 네트워크 요청을 수행할 수 없으며 네트워크 호출을 시도하는 쿼리는 무기한 중단됩니다. 아웃바운드 네트워크 액세스를 사용하도록 설정하려면 공개 미리 보기 기능을 사용하도록 설정하여 작업 영역의 미리 보기 페이지에서서버리스 SQL Warehouse의 격리된 워크로드에 대해 네트워킹을 사용하도록 설정합니다. 그렇지 않으면 네트워크 액세스가 필요한 UDF에 대해 서버리스 컴퓨팅 또는 클래식 컴퓨팅을 사용합니다.