[PySpark] Spark의 Dataframe API를 알아보자!(2)

🔊 해당 포스팅은 스파크 완벽 가이드 책과 인프런의 스파크 머신러닝 완벽 가이드 강의로 공부한 후 배운 내용을 저만의 방식으로 재구성한 것임을 알립니다. 특히, 참고한 인프런 강의의 강의 자료를 그대로 사용하지 않았음을 필히 알립니다!

 

Apache Spark를 Python으로 이용하는 PySpark에 대해 알아보자

 

---

저번 포스팅에 이어 PySpark의 dataframe API 사용 방법에 대해 연이어 알아보자.

1. 조건에 맞는 데이터만 추출하기

특정 조건에 맞는 데이터를 추출하기 위해 사용하는 메소드로는 filter() 메소드가 존재한다. 이 때, filter() 메소드에는 스파크 데이터프레임의 Column Type 객체를 넣어주어야 한다. 여기서 Column Type 객체란게 잘 와닿지 않을 텐데, 우리가 저번 시간에 select() 메소드를 활용해서 특정 칼럼을 추출할 때, 아래와 같은 5가지 방법을 사용한다고 했다.

 

spark_df.select('Age').limit(3).show()
spark_df.select(['Age']).limit(3).show()
spark_df.select(spark_df['Age']).limit(3).show()
spark_df.select(spark_df.Age).limit(3).show()
spark_df.select(col('Age')).limit(3).show()

 

위 방법들 중 1,2번째 방법은 Column Type 객체가 아니고 3,4,5 번째 방법은 Column Type 객체를 의미한다. 혹시 모르니 3,4,5 번째 방법을 따로 코드로 type을 확인해보면 아래와 같다.

 

 

그림을 보면 pyspark.sql.column.Column 이라는 타입이라고 나온다. 다시 filter() 메소드로 돌아와서, filter() 메소드 안에는 위처럼 Column 타입의 형태로 조건을 명시해주어야 한다. 아래처럼 말이다.

 

path = '/Users/younghun/Desktop/gitrepo/data/titanic/train.csv'

spark_df = spark.read.csv(path, header=True, inferSchema=True)

# filter 메소드안에는 Column type을 넣어주어야 함!
spark_df.filter(col('Age') > 25).limit(5).show()
spark_df.filter(spark_df['Age'] > 25).limit(5).show()
spark_df.filter(spark_df.Age > 25).limit(5).show()

# error 발생!
spark_df.filter('Age' > 25).limit(5).show()

 

그리고 위 filter() 메소드는 where() 메소드로 똑같이 대체할 수 있다. 여기서 where() 은 쉽게 말하면 SQL에서 자주 사용하는 WHERE 조건문 때의 키워드 where() 이라고 생각하면 된다.

참고로 filter(), where() 메소드 둘 다 위와 같이 Column Type 객체처럼 명시하지 않고 쌍따옴표로 감싸서 SQL 쿼리 형태로 명시해주어도 정상 동작한다. 아래처럼 말이다.

 

spark_df.filter("Age > 25").limit(5).show()
spark_df.where("Age > 25").limit(5).show()

 

그리고 조건을 두 가지 이상 명시하면서 AND, OR 과 같은 논리연산자는 판다스처럼 각각 &, | 를 사용하면 된다. 사용법은 아래와 같다.

 

from pyspark.sql.functions import col

cond1 = col('Age') > 25
cond2 = col('Embarked') != 'S'

spark_df.filter(cond1 & cond2).limit(5).show()
spark_df.where(cond1 | cond2).limit(5).show()

 

추가적으로 Column Type 객체를 명시하면서 SQL의 LIKE 구문 같은 것들을 추가 메소드로 활용할 수 있다.

 

spark_df.filter(col('Name').like('%Mr.%')).limit(5).show()
spark_df.where("Name LIKE '%Mr.%'").limit(5).show()  # SQL 쿼리 형태로 명시도 가능!

2. 특정 칼럼 값을 기준으로 데이터 정렬하기

이번엔 스파크의 데이터프레임에서 특정 칼럼 값에 따라 데이터 Row를 오름(또는 내림)차순 정렬하는 방법을 알아보자. 크게 orderBy() 메소드, sort() 메소드가 있는데, 사용 방법은 동일하다. 먼저 orderBy() 메소드 사용 예시이다. sort() 메소드는 아래의 orderBy() 를 대체하기만 하면 된다.

 

# orderBy -> 단일 칼럼 기준으로 정렬
spark_df.orderBy("PassengerId", ascending=False).limit(5).show()
spark_df.orderBy(col('PassengerId'), ascending=False).limit(5).show()
spark_df.orderBy(col('PassengerId').desc()).limit(5).show()

# orderBy -> 복수 칼럼 기준으로 정렬
spark_df.orderBy("PassengerId", "Age", ascending=[True, False]).limit(5).show()
spark_df.orderBy(col("PassengerId"), col("Age"), ascending=[True, False]).limit(5).show()
spark_df.orderBy(col("PassengerId").asc(), col("Age").desc()).limit(5).show()

 

참고로 판다스에서도 sort_values() 메소드에서 위와 같이 복수 칼럼 기준일 때, 각 칼럼마다 오름차순, 내림차순 정렬할 수 있는지는 처음알았다. 아래처럼 사용하면 된다.

 

pandas_df.sort_values(by=['Pclass', 'Name'], ascending=[True, False]).head()

 

그러면 지금까지 배워본 내용들을 살짝 응용해서 아래와 같은 코드를 짜볼 수 있다.

 

select_cols = ["PassengerId", "Survived", "Name", "Age", "Sex"]
filter_cond = col("Age") > 35

spark_df.select(*select_cols) \
        .filter(filter_cond) \
        .orderBy("Age", "PassengerId", ascending=[False, True]) \
        .show()

3. 특정 칼럼으로 데이터 그룹핑하기

이번엔 groupBy() 라는 메소드를 활용해서 특정 칼럼을 기준으로 데이터를 그룹핑시켜보자. 판다스의 groupby() 메소드와 마찬가지로 그룹핑할 칼럼을 지정해주고 그리고 집계할 함수에 집계할 칼럼을 명시해주면 된다. 단, 집계할 칼럼을 명시해줄 때는 문자열 형태로 칼럼 이름을 넣어주도록 하자. 그렇지 않으면 에러가 발생한다. 그리고 만약 집계할 칼럼 이름을 명시해주지 않으면 모든 칼럼에 대해 집계를 해버리게 된다.(단, count() 메소드는 칼럼 이름을 명시해주지 않아도 행의 개수만 딱 반환하도록 되어 있다)

 

spark_df.groupBy("Survived").max("Age").show()
spark_df.groupBy(col("Survived")).max("Age").show()
spark_df.groupBy(spark_df.Survived).max().show()  # 모든 칼럼에 대해 max 집계 됨

 

이번엔 여러 칼럼을 기준으로 그룹핑해보자.

 

# 여러 칼럼을 기준으로 그룹핑
spark_df.groupBy("Sex", "Survived").max("Age").show()
spark_df.groupBy(["Sex", "Survived"]).max("Age").show()  # 리스트 형태로 넣어도 정상동작함

 

그렇다면 집계할 칼럼이 여러가지인데, 각 칼럼마다 서로 다른 집계 함수를 사용하고 싶을 땐 어떻게 할까? 그 때는 agg() 메소드를 사용하면 된다.

 

spark_df.groupBy(col("Sex"), col("Survived")) \
        .agg(max("Age").alias("Max_Age"),
            min("Age").alias("Min_Age"),
            avg("Fare").alias("Avg_Fare")) \
        .show()

4. 특정 칼럼을 내 맘대로 가지고 놀기

이번에는 데이터프레임이 기존에 갖고 있던 칼럼(들)을 활용하는 방법에 대해 알아보자. 대표적으로는 withColumn() 이라는 메소드를 사용할텐데, 이 메소드는 크게 3가지 역할을 한다. 첫 번째는 기존 칼럼으로부터 어떠한 '변환'을 취해서 새로운 칼럼을 생성하는 것, 두 번째는 기존 칼럼값을 변경하는 것, 세 번째는 기존 칼럼 값 타입을 바꾸는 것이다.

 

그런데 withColumn() 메소드를 사용할 때는 새롭게 만들 칼럼 이름 즉, 첫 번째 인자에는 무조건 문자열 형태로 입력해주어야 하고, 기존 칼럼을 넣는 두 번째 인자에는 반드시 col('column') 또는 df['column'] , df.column 과 같이 Column Type 형태로 넣어주어야 함을 잊지 말자.

 

가장 먼저 새로운 칼럼을 생성해보도록 하자.

 

from pyspark.sql.functions import avg

# 스파크에서 데이터프레임 복사본을 만들기 위해서는 아래와 같이 수행
copy_df = spark_df.select("*")

# Age에다가 평균나이를 더해보자
avg_age = copy_df.select(avg(col("Age"))).first()[0]
new_age_df = copy_df.withColumn("New_Age", col("Age") + avg_age)
new_age_df.limit(5).show()

 

두 번째는 기존 칼럼값을 변경하는 방법인데, 처음엔 아래와 같이 동작할 거라고 생각할 수 있다. 하지만 아래 코드는 에러가 발생한다.

 

copy_df = spark_df.withColumn("Age", 0)
copy_df.show()

 

이 때, 에러 메세지가 col should be Column 이라고 나오게 되는데, 이것은 위에서 0 값을 Column Type 으로 만들어주어야 한다는 의미이다. 이렇게 상수값 또는 문자열을 Column Type 으로 만들어주기 위해서는 pyspark.sql.functions 의 lit 이라는 메소드를 임포트해서 사용해야 한다. 여기서 lit 은 literal 이라는 단어의 줄임말이다.

 

from pyspark.sql.functions import lit

copy_df = spark_df.withColumn("Age", lit(0))
copy_df.show()

 

그 이외에 칼럼 값이 문자열일 때, 해당 문자열 일부를 가져오는 substring 메소드나 특정 구분자를 기준으로 쪼개서 가져오는 split 메소드도 존재한다.

 

다음은 특정 칼럼의 데이터 타입을 변경하는 방법이다.

 

copy_df = spark_df.withColumn("Age", col("Age").cast("integer"))
copy_df.show()

5. 특정 칼럼 또는 Row를 삭제하기

특정 칼럼을 삭제하는 방법은 drop() 메소드를 활용하면 되므로 간단하다. 칼럼명은 문자열 형태로 넣어주거나 Column Type 형태로 명시해주어도 된다. 하지만 한 번에 여러개의 칼럼을 삭제할 때는 여러 칼럼을 문자열 형태로 열거하는 식으로 해주어야만 한다. 리스트에 넣어주거나 또는 Column Type 형태로 명시해주면 에러가 발생한다.

 

# 특정 칼럼을 삭제하기
spark_df.drop("Pclass").show()
spark_df.drop(col("Pclass")).show()

# 한 번에 여러개 칼럼 삭제하기
spark_df.drop("Pclass", "Name").show()

# 아래의 두 경우는 error 발생
spark_df.drop(["Pclass", "Name"]).show()  
spark_df.drop(col("Pclass"), col("Name")).show()

 

다음은 결측치가 있는 Row를 삭제하는 방법에 대해 알아보자. 대표적으로 dropna() 메소드와 na.drop() 메소드가 존재한다. 두 메소드 모두 subset 이라는 인자에 문자열(하나의 칼럼일 경우), 튜플 또는 리스트 형태로 넣어줄 수 있는데, 이 subset 은 Row를 삭제할 때, 어느 칼럼에 결측치가 있는 Row를 삭제할지를 결정하는 기준이 된다.

 

만일 아래와 같은 코드가 있다면, Age, Embarked 칼럼 값이 모두 결측치인 Row들만 제거하라는 의미이다.

 

spark_df.dropna(subset=["Age", "Embarked"]).show()

6. 결측치가 있는 Row들 체크하기

다음은 결측치가 있는 Row들을 체크하는 방법이다. 판다스의 경우 isnull().sum() 과 같은 매우 편한 메소드가 있지만, 스파크에서는 이렇게 할 수 없다. 특정 칼럼마다 결측치가 있는지 없는지 체크하는 메소드인 isNull() 메소드 또는 isnan() 메소드를 활용해야 한다. 먼저 isNull() 메소드를 사용하는 방법은 아래와 같다.

 

spark_df.filter(col("Age").isNull()).show()

 

반면에, isnan() 메소드는 pyspark.sql.functions 클래스에서 임포트해서 아래처럼 사용해야 한다.

 

from pyspark.sql.functions import isnan

spark_df.where(isnan(col("Age"))).show()

 

그런데, 여기서 스파크에서 Null 과 NaN 값 둘 다 결측치인지 아니면 Null만 결측치인지 헷갈릴 수 있다. 결론은 최신 스파크 버전에서는 Null 값만을 결측치로 간주하므로 NaN 값을 마주하게 되면 자연스레 Null 값으로 바꾸어버린다.

 

NaN은 구체적으로 Not a Number의 줄임말로, 파이썬의 None 타입을 넘파이 형태로 만들어 버리면 NaN 값이 되버린다.

 

import numpy as np
import pandas as pd

data = {"name": ["jo", "lee"],
       "age": [28, None]}
pd.DataFrame(data)

 

위 결과를 보면 None 값이 NaN 값으로 판다스 데이터프레임에 들어가 있는 것을 볼 수 있다.

 

다음은 각 칼럼별로 결측치 개수가 몇 개가 존재하는지 확인하는 방법이다. 이 때, 활용하는 메소드는 행 개수를 세는 count() 와 SQL에서의 CASE WHEN 역할을 하는 when() 메소드이다.

 

from pyspark.sql.functions import count, when

cond = [count(when(col(c).isNull(), c)).alias(f"{c}_null_cnt") for c in spark_df.columns]
spark_df.select(cond).show()

 

원래 count() 메소드는 Null인 Row를 카운트하지 않는 것이 정석이다. 그런데 위에서 결측치 개수를 어떻게 셀 수 있을까 하는 의문이 들 수 있다.(그래서 필자가 질문을 하기도 했다..) 이유는 when() 메소드의 특성 때문인데, when() 메소드 안에서 c 라는 칼럼이 결측치 일 때, c 라는 값으로 대체하라는 의미인데, 이 대체하는 c 라는 값이 칼럼 c 에 들어있는 값(value)을 의미하는 것이 아닌 칼럼 이름 c 문자열 자체 값을 의미한다. 따라서 when() 메소드는 실질적으로 c 라는 칼럼에 결측치가 있는 값들을 'c' 문자열로 간주한 후, 문자열이 c 인 Row 개수를 셈으로써 결측치 개수를 셀 수 있게 되는 것이다.

7. 결측치 대체하기

다음은 결측치를 어떤 값으로 대체하는 방법이다. 크게 fillna() 메소드와 na.fill() 메소드를 사용할 수 있다. 두 메소드 모두 value 인자에 대체할 값을 넣어줄 수 있고 subset 인자로 결측치를 대체할 칼럼이 무엇인지 넣어줄 수 있다. 인자로 아무것도 넣지 않으면 모든 칼럼의 결측치에 대체할 값을 넣어준다.

 

참고로 value 인자에 정수형으로 넣어주면 정수형 타입을 갖고 있는 칼럼들의 결측치를 알아서 찾아서 대체해주고 value 인자에 문자열을 넣어주면 문자열 타입을 갖고 있는 칼럼들의 결측치를 알아서 찾아 대체해준다.

 

from pyspark.sql.functions import avg

avg_age = spark_df.select(avg("Age")).first()[0]

spark_df.fillna(value=avg_age, subset="Age").show()
spark_df.na.fill(value=avg_age, subset=["Age"]).show()

 

또 결측치가 존재하는 칼럼에 맞게 서로 다른 대체값을 부여하려면 딕셔너리 형태로 정의해서 부여해줘도 된다.

 

from pyspark.sql.functions import avg

avg_age = spark_df.select(avg("Age")).first()[0]
embarked = "ZZZ"

dictionary = {"Age": avg_age, "Embarked": embarked}

spark_df.fillna(value=dictionary).show()

8. 일반 UDF를 스파크 UDF로 변환 후 적용하기

일반 UDF란, 어려운 것이 아니라 우리가 일반적으로 파이썬으로 정의하는 '일반' 함수를 의미한다. 판다스에서는 이렇게 일반 UDF를 활용해서 판다스의 데이터프레임 또는 Series에 apply,lambda 함수를 활용해서 적용해 본 적이 있다. 이와 비슷하게 스파크에서도 UDF를 활용할 수 있는데, 사용 방법은 아래와 같다.

 

우선, 일반 UDF 함수를 정의하자. 해당 예시에서는 아래의 UDF를 사용하려 한다.

 

def change_gender(gender):
    if gender is None:
        return None
    if gender == 'male':
        return 'super_male'
    else:
        return 'super_female'

 

다음은 udf 메소드를 따로 임포트해서 일반 UDF를 스파크용 UDF로 변환해주어야 한다. udf() 메소드를 활용하는 방법은 다음과 같다. udf(lambda x: 일반_udf, return_type) 인데, 개인적으로 특이하게 동적 언어인 파이썬에서 미리 함수 반환 타입을 정의해주는 점이 특이하다고 느꼈다.

 

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

# spark UDF로 변환
spark_udf = udf(lambda x: change_gender(x), StringType())
# spark UDF를 spark dataframe에 적용
spark_df.withColumn("New_Gender", spark_udf(col("Sex"))).show()