사용자 정의 스칼라 함수 - Scala

아티클
10/26/2024

이 문서에는 Scala UDF(사용자 정의 함수) 예제가 포함되어 있습니다. UDF를 등록하는 방법, UDF를 호출하는 방법, Spark SQL의 하위 식 평가 순서와 관련된 주의 사항을 보여 줍니다. 자세한 내용은 외부 사용자 정의 스칼라 함수(UDF)를 참조하세요.

참고 항목

공유 액세스 모드를 사용하는 Unity 카탈로그 지원 컴퓨팅 리소스의 Scala UDF에는 Databricks Runtime 14.2 이상이 필요합니다.

UDF로 함수 등록

val squared = (s: Long) => {
  s * s
}
spark.udf.register("square", squared)

Spark SQL에서 UDF 호출

spark.range(1, 20).createOrReplaceTempView("test")

%sql select id, square(id) as id_squared from test

DataFrames에서 UDF 사용

import org.apache.spark.sql.functions.{col, udf}
val squared = udf((s: Long) => s * s)
display(spark.range(1, 20).select(squared(col("id")) as "id_squared"))

평가 순서 및 Null 검사

Spark SQL(SQL과 DataFrame 및 데이터 세트 API 포함)은 하위 식의 평가 순서를 보장하지 않습니다. 특히 연산자 또는 함수 입력이 반드시 왼쪽에서 오른쪽 또는 다른 고정 순서로 평가되는 것은 아닙니다. 예를 들어 논리적 AND 및 OR 식에는 왼쪽에서 오른쪽 “단락” 의미 체계가 없습니다.

따라서 쿼리 최적화 및 계획 중에 식과 절의 순서가 변경될 수 있으므로 부울 식의 부작용 또는 평가 순서와 WHERE 및 HAVING 절의 순서에 의존하는 것은 위험합니다. 특히 UDF가 SQL의 단락 의미 체계를 null 검사에 사용하는 경우 UDF를 호출하기 전에 Null 검사가 수행된다는 보장이 없습니다. 예를 들면 다음과 같습니다.

spark.udf.register("strlen", (s: String) => s.length)
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", (s: String) => if (s != null) s.length else -1)
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

형식화된 데이터 세트 API

참고 항목

이 기능은 Databricks Runtime 15.4 이상의 공유 액세스 모드를 사용하는 Unity 카탈로그 지원 클러스터에서 지원됩니다.

형식화된 데이터 세트 API를 사용하면 사용자 정의 함수를 사용하여 결과 데이터 세트에 대한 맵, 필터 및 집계와 같은 변환을 실행할 수 있습니다.

예를 들어 다음 Scala 애플리케이션은 map() API를 사용하여 결과 열의 숫자를 접두사 문자열로 수정합니다.

spark.range(3).map(f => s"row-$f").show()

이 예제에서는 map() API를 사용하지만, 이 API는 형식화된 다른 데이터 세트 API(예: filter(), mapPartitions(), foreach(), foreachPartition(), reduce() 및 flatMap())에도 적용됩니다.

다음을 통해 공유