다음을 통해 공유

UDF(사용자 정의 함수)란?

  • 아티클

UDF(사용자 정의 함수)를 사용하면 Azure Databricks에서 기본 제공 기능을 확장하는 코드를 다시 사용하고 공유할 수 있습니다. UDF를 사용하여 복잡한 계산, 변환 또는 사용자 지정 데이터 조작과 같은 특정 작업을 수행합니다.

참고 항목

공유 액세스 모드가 있는 클러스터에서 Python 스칼라 UDF는 Databricks Runtime 13.3 LTS 이상에서 지원되는 반면 Scala UDF는 Databricks Runtime 14.2 이상에서 지원됩니다.

Databricks Runtime 13.3 LTS 이상의 SQL 구문을 사용하여 Unity Catalog에 Python 스칼라 UDF를 등록할 수 있습니다. Unity에서 사용자 정의 함수(UDF) 을 참조하세요 Catalog.

UDF는 언제 사용해야 하나요?

기본 제공 Apache Spark 함수로 표현하기 어려운 논리에 UDF를 사용합니다. 기본 제공 Apache Spark 함수는 분산 처리에 최적화되어 있으며 일반적으로 대규모로 더 나은 성능을 제공합니다. 자세한 내용은 함수를 참조하세요.

Databricks는 임시 쿼리, 수동 데이터 정리, 예비 데이터 분석 및 중소 규모의 데이터 세트에 대한 작업에 UDF를 권장합니다. UDF의 일반적인 사용 사례에는 데이터 암호화 및 암호 해독, 해시, JSON 구문 분석 및 유효성 검사가 포함됩니다.

매우 큰 데이터 세트 및 ETL 작업 및 스트리밍 작업을 포함하여 정기적으로 또는 지속적으로 실행되는 모든 워크로드에 대해 Apache Spark 메서드를 사용합니다.

등록된 UDF 및 세션 범위 UDF

SQL을 사용하여 만든 UDF는 Unity Catalog에 등록되어 관련 권한이 있는 반면, Notebook 내에서 만든 UDF는 세션 기반이며 현재 SparkSession으로 범위가 설정됩니다.

Azure Databricks에서 지원하는 모든 언어를 사용하여 세션 기반 UDF를 정의하고 액세스할 수 있습니다. UDF는 스칼라 또는 비 스칼라일 수 있습니다.

참고 항목

Unity Catalog에 등록된 SQL 및 Python 스칼라 UDF만 현재 DBSQL에서 사용할 수 있습니다.

스칼라 UDF

스칼라 UDF는 단일 행에서 작동하고 각 행에 대해 단일 값을 반환합니다. 다음 예제에서는 스칼라 UDF를 사용하여 namecolumn 각 이름의 길이를 구하고 새 columnname_length에 해당 값을 추가합니다.

+-------+-------+
| name  | score |
+-------+-------+
| alice |  10.0 |
| bob   |  20.0 |
| carol |  30.0 |
| dave  |  40.0 |
| eve   |  50.0 |
+-------+-------+

-- Create a SQL UDF for name length
CREATE OR REPLACE FUNCTION get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);

-- Use the UDF in a SQL query
SELECT name, get_name_length(name) AS name_length
FROM your_table;

+-------+-------+-------------+
| name  | score | name_length |
+-------+-------+-------------+
| alice |  10.0 |      5      |
|  bob  |  20.0 |      3      |
| carol |  30.0 |      5      |
| dave  |  40.0 |      4      |
|  eve  |  50.0 |      3      |
+-------+-------+-------------+

PySpark를 사용하여 Databricks Notebook에서 이를 구현하려면 다음을 수행합니다.

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

@udf(returnType=IntegerType())
def get_name_length(name):
   return len(name)

df = df.withColumn("name_length", get_name_length(df.name))

# Show the result
display(df)

자세한 내용은 Unity UDF(사용자 정의 함수) 및 사용자 정의 스칼라 함수인 Python참조하세요.

사용자 정의 집계 함수 (UDAFs)

UDF(사용자 정의 집계 함수)는 여러 행에서 작동하고 단일 집계 결과를 반환합니다. 다음 예제에서는 점수를 집계하는 UDAF가 정의됩니다.

from pyspark.sql.functions import pandas_udf
from pyspark.sql import SparkSession
import pandas as pd

# Define a pandas UDF for aggregating scores
@pandas_udf("int")
def total_score_udf(scores: pd.Series) -> int:
    return scores.sum()

# Group by name length and aggregate
result_df = (df.groupBy("name_length")
              .agg(total_score_udf(df["score"]).alias("total_score")))

display(result_df)

+-------------+-------------+
| name_length | total_score |
+-------------+-------------+
|      3      |     70.0    |
|      4      |     40.0    |
|      5      |     40.0    |
+-------------+-------------+

Python 및 사용자 정의 집계 함수인 Scala에 대한 pandas 사용자 정의 함수를 참조하세요.

Python UDTF(사용자 정의 table 함수)

Important

이 기능은 공개 미리 보기 상태입니다.

참고 항목

Python UDF는 Databricks Runtime 14.3 LTS 이상에서 사용할 수 있습니다.

Python UDF(사용자 정의 table 함수)는 각 입력 행에 대해 여러 행과 columns 반환할 수 있습니다. 다음 예제에서 점수 column의 각 값은 범주 list에 해당합니다. UDTF는 쉼표로 구분된 list을 여러 행으로 분할하기 위해 정의됩니다. Python 사용자 정의 table 함수(UDTF) 참조하세요.

+-------+-------+-----------------+
| name  | score |   categories    |
+-------+-------+-----------------+
| alice |  10.0 |  math,science   |
|  bob  |  20.0 |  history,math   |
| carol |  30.0 | science,history |
| dave  |  40.0 |    math,art     |
|  eve  |  50.0 |  science,art    |
+-------+-------+-----------------+


from pyspark.sql.functions import udtf

@udtf(returnType="score: int, categories: string, name: string")
class ScoreCategoriesUDTF:
    def eval(self, name: str, score: float, categories: str):
        category_list = categories.split(',')
        for category in category_list:
            yield (name, score, category)

# Apply the UDTF
result_df = df.select(ScoreCategoriesUDTF(df.score, df.categories, df.name))
display(result_df)

+-------+-------+----------+
| name  | score | category |
+-------+-------+----------+
| alice |  10.0 |   math   |
| alice |  10.0 | science  |
|  bob  |  20.0 | history  |
|  bob  |  20.0 |   math   |
| carol |  30.0 | science  |
| carol |  30.0 | history  |
| dave  |  40.0 |   math   |
| dave  |  40.0 |   art    |
|  eve  |  50.0 | science  |
|  eve  |  50.0 |   art    |
+-------+-------+----------+

성능 고려 사항

  • 기본 제공 함수 및 SQL UDF는 사용 가능한 가장 효율적인 옵션입니다.
  • Scala UDF 는 일반적으로 JVM(Java Virtual Machine) 내에서 실행되고 JVM 안팎으로 데이터를 이동하는 오버헤드를 방지하므로 더 빠릅니다.
  • Python UDF 및 Pandas UDF 는 데이터를 직렬화하고 JVM에서 Python 인터프리터로 이동해야 하기 때문에 Scala UDF보다 느린 경향이 있습니다. Pandas UDF는 Apache Arrow를 사용하여 직렬화 비용을 줄이기 때문에 Python UDF보다 최대 100배 더 빠릅니다.