Spark UDF와 explode 기능 Colab에서 실습하기 - TIL230705

📚 KDT WEEK 14 DAY 3 TIL

• UDF
• UDAF
• Explode

 

---

⚒️ UDF - User Defined Function

DataFrame이나 SQL에서 적용할 수 있는 사용자 정의 함수

• Scalar 함수 : UPPER, LOWER ...
• Aggregation 함수(UDAF) : SUM, MIN, MAX

 

Google Colab에서 실습을 진행한다.

 

pyspark와 py4j 환경설치

!pip install pyspark==3.3.1 py4j==0.10.9.5

 

SparkSession 생성

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark UDF") \
    .getOrCreate()

 

간단한 데이터프레임 생성

columns = ["Seqno","Name"]
data = [("1", "john jones"),
    ("2", "tracey smith"),
    ("3", "amy sanders")]

df = spark.createDataFrame(data=data,schema=columns)

df.show(truncate=False)

 

---

 

🟥 UDF 실습

 

pyspark에서 udf를 사용할 때는 pyspark.sql.functions를 사용한다.

 

1. 람다함수 UDF

 

# 람다함수 UDF
# F.udf에 람다함수 넣기

import pyspark.sql.functions as F
from pyspark.sql.types import *

upperUDF = F.udf(lambda z:z.upper())

df.withColumn("Curated Name", upperUDF("Name")) \
  .show(truncate=False)

 

F.udf(람다함수) 형태로 저장해서 파이썬에서 함수를 쓰듯 사용한다.

withColumn은 새로운 컬럼을 추가하는 기능으로써 upperUDF 함수의 결과를 Curated Name 컬럼에 추가하게 된다.

 

 

2. 파이썬 함수

 

# 파이썬 함수 UDF
# F.udf에 파이썬함수 넣기

def upper_udf(s):
    return s.upper()
    
upperUDF = F.udf(upper_udf, StringType())

df.withColumn("Curated Name", upperUDF("Name")) \
   .show(truncate=False)

 

일반적으로 함수를 정의해서 이를 F.udf에 넣어주는 형태이다.

람다함수와 비슷하지만 더 명확하고 이해하기 쉬운 형태인 것 같다.

 

 

 

추가로 withColumn으로 컬럼을 매번 추가하지 않고, SELECT에서 바로 원하는 컬럼과 같이 볼 수도 있다.

 

Name 컬럼과 Curated Name 컬럼이 출력

 

3. Pandas UDF

 

함수를 레코드 집합 단위로 처리하기 때문에 다른 함수보다 효율적인 방법이다.

 

# 판다스 udf
# 이 방법은 레코드 집합이 들어와서 처리하므로 효율적인 방법이다

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

# Define the UDF
# 입력타입 -> 출력타입 형태로 정의한다.
@pandas_udf(StringType())
def upper_udf_f(s: pd.Series) -> pd.Series:
    return s.str.upper()
    
# 위에서 정의한 파이썬 upper 함수를 그대로 사용
# SQL 쿼리에서 사용할 수 있도록 첫 번째 인자에 사용할 이름을 만들어준다.
upperUDF = spark.udf.register("upper_udf", upper_udf_f)
spark.sql("SELECT upper_udf('aBcD')").show()

 

spark.udf.register로 SQL 쿼리안에서 사용할 함수 이름을 만들고, spark.sql안에서 UDF를 사용했다.

 

 

 

---

 

🟦 UDAF 실습

UDF 함수로 Aggregation(SUM, AVERAGE, MIN.. 등)을 사용하는 형태이다.

 

Pandas UDF를 사용한다.

# dataframe에 User Define Aggregation Fuction 사용해보기

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

# Define the UDF
@pandas_udf(FloatType())
# pd.Series를 입력으로 받아 float 값을 출력으로 냄
def average_udf_f(v: pd.Series) -> float:
    return v.mean()

averageUDF = spark.udf.register('average_udf', average_udf_f)
spark.sql('SELECT average_udf(a) FROM test').show() # a의 평균값 계산하기 (a로 그룹하여 평균값 계산)

 

위에서 UDF를 사용한 형태와 비슷한 형태이다.

하지만, Aggregation 함수는 withColumn을 사용하지 못한다고 한다.

 

다른 기능으로, 데이터프레임에 .agg를 사용해서 값을 확인할 수 있다.

 

averageUDF를 사용한 모습

 

++) 기존의 함수와 함께 내가 만든 UDF를 확인할 수 있는 방법이 있다.

for f in spark.catalog.listFunctions():
    print(f[0])

 

여러 함수 중에서 내가 만든 함수 average_udf가 보인다.

 

---

 

🟩 Explode

 

이름별로 사용하는 프로그래밍 언어와, 그 사람의 특징을 넣어둔 간단한 데이터를 주었다.

사용하는 언어는 여러 개가 있을 수 있기 때문에 한 컬럼에 리스트로 저장하게 되었다.

이 언어들을 하나의 레코드로 쪼개서 별개의 레코드로 생성(explode)해보자.

 

arrayData = [
        ('James',['Java','Scala'],{'hair':'black','eye':'brown'}),
        ('Michael',['Spark','Java',None],{'hair':'brown','eye':None}),
        ('Robert',['CSharp',''],{'hair':'red','eye':''}),
        ('Washington',None,None),
        ('Jefferson',['1','2'],{})]

df = spark.createDataFrame(data=arrayData, schema = ['name','knownLanguages','properties'])
df.show()

 

 

데이터를 보면, 한 컬럼에 리스트 형태의 데이터가 삽입된 것을 알 수 있다.

 

여기서 explode를 사용하면 리스트의 각 데이터별로 레코드를 생성할 수 있다.

# knownLanguages 필드를 언어별로 잘라서 새로운 레코드로 생성
from pyspark.sql.functions import explode
df2 = df.select(df.name,explode(df.knownLanguages)) # knownLanguages 를 explode한 컬럼 + name 컬럼으로 select
df2.printSchema()
df2.show()

 

Name 컬럼이 중복되고, 언어별로 레코드가 생성되었다!

 

---

 

이번에는 Dictionary 형태로 저장된 데이터를 구조(struct)로 저장하여 불러오는 실습을 해보자.

 

여기서는 따로 저장한 orders.csv 파일을 사용했다.

 

orders.csv의 일부 내용

 

items 컬럼에 name, quantity, id별로 각 value가 저장된 형태이다.

이 구조를 명확하게 명시해서 각각 Key를 불러왔을 때 value가 나타나도록 코딩해보자.

 

먼저 csv 파일을 불러온다.

from pyspark.sql.functions import *
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, LongType

order = spark.read.options(delimiter='\t').option("header","true").csv("orders.csv") # 탭으로 구분된 파일이라 \t 사용

 

여기서 printSchema를 사용해서 타입을 확인해보면, 아직 구조없이 String을 가지고 있음을 알 수 있다.

 

아직은 items 컬럼에 구조가 없다.

 

ArrayType을 사용해서 구조를 하나 만들어준다.

 

# items 컬럼의 구조 잡기
# 데이터프레임을 이용해서 해보기
struct = ArrayType(
    StructType([
        StructField("name", StringType()),
        StructField("id", StringType()),
        StructField("quantity", LongType())
    ])
)

# items 형태를 고려해서 from_json + 아까 만든 items 구조를 활용해서 explode
order.withColumn("item", explode(from_json("items", struct))).show(truncate=False)

 

items안의 형태대로 구조를 정의하고, explode에 from_json(컬럼이름, 사용할 구조)를 넣어 구조를 넣었다.

 

withColumn을 사용했기 때문에, 기존의 컬럼은 남고 추가 컬럼이 붙었다.

 

새로 추가된 item 컬럼에 value 값만 남아 저장되었다.

이제 필요없는 기존 컬럼 items를 drop하고, 바뀐 스키마를 확인해보자.

 

# items 컬럼 제거
order_items = order.withColumn("item", explode(from_json("items", struct))).drop("items")

 

item 컬럼에 구조가 들어간 모습!

 

이 구조를 테이블에서 활용해보자!

 

# 테이블로 만들기
order_items.createOrReplaceTempView("order_items")

# 쿼리문
# item의 quantity를 사용하려면 item.quantity로 사용하면 된다!
spark.sql("""
    SELECT order_id, CAST(average_udf(item.quantity) as decimal) avg_count
    FROM order_items
    GROUP BY 1
    ORDER BY 2 DESC""").show(5)

 

쿼리문의 결과