[SparkSQL] UDF 개념 및 코드

 

 

UDF

 

User Define Function. 즉, 사용자 지정 함수를 말한다.

앞선 포스팅에서 데이터 프레임을 처리하는 여러 함수들을 보았다.

SparkSQL에선 사용자가 직접 어떤 작업을 수행하는 함수를 명명하고, Spark에 등록하여 쓸 수 있다.

자세하게 알아보자. 

 

 

"본 포스팅은 패스트캠퍼스의 강의를 듣고, 정리한 자료임을 밝힙니다."

 

 

 

 

Basic Setting

 

import os
import findspark
findspark.init(os.environ.get("SPARK_HOME"))
import pyspark
from pyspark import SparkConf, SparkContext
import pandas as pd
import faulthandler
faulthandler.enable()
from pyspark.sql import SparkSession

spark = SparkSession.builder.master('local').appName("dataframe").getOrCreate()

# Data

stocks = [
    ('Google', 'GOOGL', 'USA', 2984, 'USD'), 
    ('Netflix', 'NFLX', 'USA', 645, 'USD'),
    ('Amazon', 'AMZN', 'USA', 3518, 'USD'),
    ('Tesla', 'TSLA', 'USA', 1222, 'USD'),
    ('Tencent', '0700', 'Hong Kong', 483, 'HKD'),
    ('Toyota', '7203', 'Japan', 2006, 'JPY'),
    ('Samsung', '005930', 'Korea', 70600, 'KRW'),
    ('Kakao', '035720', 'Korea', 125000, 'KRW'),
]


# Schema

stockSchema = ['name', 'ticker', 'country', 'price', 'currency']


# createDataFrame()
df = spark.createDataFrame(data = stocks, schema=stockSchema)

# createOrReplaceTempView()
df.createOrReplaceTempView("stock")

 

 

 

 

UDF

---

사용자가 직접 정의하는 함수를 말한다.

User Define Function.

 

 

1. spark.udf.register()

정의된 함수를 spark에서 쓸 수 있도록 등록한다.

spark.udf.register("함수 이름", 함수, return할 데이터 타입)

 

from pyspark.sql.types import LongType

def squared(n):
    return n * n

spark.udf.register("squared", squared, LongType())

 

사용하기

 

# 가격을 제곱했다.
spark.sql("select name, squared(price) from stocks").show()


+-------+--------------+
|   name|squared(price)|
+-------+--------------+
| Google|       8904256|
|Netflix|        416025|
| Amazon|      12376324|
|  Tesla|       1493284|
|Tencent|        233289|
| Toyota|       4024036|
|Samsung|    4984360000|
|  Kakao|   15625000000|
+-------+--------------+

----

# 제곱한 가격이 1,000,000을 넘지 않는 경우에만 출력하도록 했다.
spark.sql("select name, squared(price) from stocks where squared(price) < 1000000").show()


+-------+--------------+
|   name|squared(price)|
+-------+--------------+
|Netflix|        416025|
|Tencent|        233289|
+-------+--------------+

 

 

 

2. 조금 더 실용적인 함수를 만들어 보자.

 

# 통화를 한글로 변환하고, 가격이랑 합쳐주자.
def currency_ko(n):
    if n == 'USD':
        return '달러'
    elif n == 'KRW':
        return '원'
    elif n == 'JPY':
        return '엔'
    else:
        return '위안'

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


<function __main__.currency_ko(n)>

----

# SQL문의 CONCAT() 함수를 써줬다.
spark.sql("select name, concat(price, currency_ko(currency)) as price from stocks").show()


+-------+--------+
|   name|   price|
+-------+--------+
| Google|2984달러|
|Netflix| 645달러|
| Amazon|3518달러|
|  Tesla|1222달러|
|Tencent| 483위안|
| Toyota|  2006엔|
|Samsung| 70600원|
|  Kakao|125000원|
+-------+--------+

 

 

 

UDF에 대해 알아보았다.

다음 포스팅에선 SparkSQL의 백엔드 프로세싱에 대해 알아보자.

수고하셨습니다.