[Spark] 사용자 정의함수(UDF)

UDF (user defined function)

- 사용자 정의 함수는 스파크에서 가장 강력한 기능 중 하나

- 파이썬이나 스칼라 그리고 외부 라이르브러리를 사용해 사용자가 원하는 형태로 트랜스포메이션을 만들 수 있게 함

- UDF는 한개 이상의 컬럼을 입력으로 받고 반환 가능

 

UDF 는 스파크의 드라이버 프로세스내에 저장되어 운용되며, 각 익스큐터에 전달하여 처리가 됨

 

pyspark sql 구문에서 udf를 사용하려면 정의한 udf를 sql 구문내 function으로 사용할 명령어를 아래와 같이 꼭 등록해야 사용 가능

>> 파이썬 함수를 SQL 함수로 등록!!!

spark.udf.register("word_count", word_count)

### spark udf 생성
from pyspark.sql.functions import udf
from pyspark.sql.typesimport IntegerType

# 메서드로 직접 생성
slen = udf(lambda s: len(s), IntegerType())

# decorator로 생성
@udf
def to_upper(s):
	if sisnotNone:
		return s.upper()

@udf(returnType=IntegerType())
def add_one(x):
	if xisnotNone:
		return x + 1

@udf(IntegerType())
def word_count(words:str, keyword:str):
	return words.count(keyword)


### spark dataframe에서 udf 사용
df = spark.createDataFrame([(1, "John Doe", 21, "o")], ("id", "name", "age", "keyword"))
df.select(slen("name").alias("slen(name)"), to_upper("name"), add_one("age")).show()


### sql 구문에서 udf 사용
spark.udf.register("word_count", word_count)
df.createOrReplaceTempView("df_view")
sql = '''
SELECT 
	id,
	name,
	keyword, 
	word_count(name,keyword) AS word_cnt 
FROM df_view
'''

반응형

HIVE UDF 사용 가능

- 하이브 문법을 사용해서 만든 UDF/UDAF도 사용 가능

- 단, 하이브를 쓰려면 sparksession 생성과정에서 enableHiveSupport()를 명시해 반드시 하이브 지원 기능을 활성화 해야함

- 스파크 실행시에 --jars 옵션을 이용하여 jar 파일을 추가할 것.

sql("""ADD JAR hdfs:///user/sample.jar""")
sql("""CREATE TEMPORARY FUNCTION sampleudf AS 'sdk.hive.SampleUDF'""")
sql("""SELECT sampleudf('a')""")

- Create 시 temporary 키워드를 제거하면 하이브 메타스토어에 영구 함수로 등록됨