[PySpark] 타이타닉 데이터로 분류 모델 만들기

🔊 본 포스팅은 PySpark를 활용한 Kaggle Notebook을 필사하면서 배우게 된 코드 내용을 기반으로 포스팅했음을 알려드립니다. 또한 앞으로 소개될 PySpark의 문법에 대해서 생소하시다면 여기를 참고해 간단한 예시를 통해 이해를 할 수 있습니다.

 

이번 포스팅에서는 PySpark 그 중에서도 Spark SQL 과 Spark MLlib을 활용한 머신러닝 분류 모델을 만드는 방법에 대해 소개하려고 한다. 활용한 데이터는 머신러닝에 입문할 때 가장 많이 사용되는 타이타닉의 Train 데이터를 사용했다.

 

Titanic 사건은 수많은 사망자를 양산한 비극적인 실화 이야기이다.

1. 필요한 라이브러리와 데이터 로드

가장 먼저 할 일은 데이터 로드와 모델 빌딩 시 사용될 Spark SQL, Spark MLlib의 여러가지 메소드들을 임포트하자. 참고로 PySpark를 사용해 데이터를 로드할 때는 os.chdir() 로 디렉토리를 변경하고 데이터를 로드할 수가 없다는 점을 알아두자.(이유를 구글링해서 찾아보려 했는데.. spark os.chdir 등과 같은 키워드로.. 그런데 이유에 대해서는 찾지 못했습니다.. 이에 대해 아는 분이 있다면 댓글 달아주시면 매우 감사하겠습니다 (__) )

 

import pandas as pd
import numpy as np
import os
import matplotlib.pyplot as plt
import seaborn as sns
plt.rc('font', family='AppleGothic')
plt.rcParams['axes.unicode_minus'] = False
# Pyspark - SQL
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
from pyspark.sql.functions import mean, col, split, regexp_extract, when, lit

# Pyspark - ML
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import QuantileDiscretizer

# Spark의 첫 시작인 SparkSession을 만들어주자!
spark = SparkSession.builder\
        .appName('Play with pypsark ML on Titanic Data')\
        .getOrCreate()
# 데이터 로드
df = spark.read.csv('/Users/younghun/Desktop/gitrepo/data/titanic/train.csv', header=True, inferSchema=True)
# toPandas()를 이용해 판다스에서 제공하는 데이터프레임 형태로 출력
df.limit(3).toPandas()

 

3개의 데이터 미리보기

• df.limit(num=3) : 해당 데이터프레임의 미리보기 하는 방법이다. Pandas에서는 head() 와 같다고 생각하면 된다. 차이점이라고 한다면 Pandas에서 head() 는 아무런 인자를 넣지 않아도 디폴트가 5개의 데이터를 보여주지만 limit() 은 어떤 인자도 넣어주지 않는다면 에러가 발생한다.
• df.toPandas() : 데이터프레임을 Pandas의 데이터프레임 형태로 변환해준다. 따라서 새로운 변수로 할당 시키는 것이 가능하다.

2. Matplotlib, Seaborn을 활용한 간단한 시각화

Matplotlib, Seaborn 을 활용하려면 우선 Spark의 데이터프레임에서 Pandas의 데이터프레임으로 변환해주고 시각화를 구현해야 한다. 따라서 먼저 toPandas() 메소드로 Pandas 형태의 데이터프레임을 새로 할당해주자.

 

# Pandas 데이터프레임 형태로 우선 변환!
pandas_df = df.toPandas()
print("pandas_df 타입:", type(pandas_df))

# Seaborn 사용해보기
plt.figure(figsize=(10, 5))
plt.title("타이타닉 탑승객의 Age KDE 분포")
sns.distplot(pandas_df['Age'])
plt.show()

 

Age 칼럼의 Distplot 그려보기

 

위 결과화면에서 볼 수 있듯이 toPandas() 메소드로 새로 할당한 데이터프레임의 type은 pandas.dataframe 임을 확인할 수 있다.

3. PySpark로 결측치 체크하고 기존변수로부터 파생변수 생성하기

결측치를 체크하기 위해 물론 Spark의 데이터프레임 형태를 Pandas 형태의 데이터프레임 형태로 바꾼 후 평소에 자주 사용했던 Pandas API에서 제공하는 결측치 확인 메소드를 사용해도 된다. 하지만 필자가 이런 포스팅을 작성하는 목적은 PySpark에 대해 익숙해짐이 목적이기 때문에 Spark의 데이터프레임 형태에서 결측치를 체크하는 방법에 대해 알아보자. 결측치를 체크하는 방법은 크게 2가지가 있다.

 

1. isnan() : pyspark.sql.functions 라이브러리에 속한다. 사용방법은 isnan('체크할 column 이름')
2. isNull() : pyspark.sql.Column 라이브러리에 속한다. 사용방법은 col('체크할 column 이름').isNull()

# 위 2가지 방법을 동시에 사용해보자!
# 결측치가 있는 변수를 체크하고 결측치가 몇 개 있는지 살펴보기
from pyspark.sql.functions import isnan, count
df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns])\
  .show()

 

위 코드에서 when() 메소드가 등장하는데 when() 메소드는 filter() 메소드와 비슷한 기능을 한다고 볼 수 있다. when(조건A, 조건A가 True일 시 value).otherwise(조건A가 False일 시 value)로 사용한다. 참고로 조건을 여러개 넣고 싶다면 논리연산자 | , & 사용이 가능하다. 위 코드에 대한 결과화면은 다음과 같다.

 

결측치를 체크한 결과 화면

 

추가적으로 불필요한 칼럼인 Cabin 칼럼을 없애주기 위해 drop() 메소드를 사용할 수 있다. 이는 Pandas와 메소드 이름이 동일하지만 inplace = False 이기 때문에 다음과 같이 재할당을 해주어야 한다.

 

# Cabin 변수는 결측치가 50%가 넘기 때문에 해당 변수를 삭제하자
df = df.drop('Cabin')

 

탑승객의 이름을 담고 있는 Initial 칼럼을 이용해 파생변수를 만들어보자. withColumn() 메소드를 사용하면 되는데, 여기서는 PySpark에서 제공하는 정규표현식 API를 함께 이용해보자. 우선 각 메소드에 대한 사용방법을 설명하면 다음과 같다.

 

• df.withColumn('New column', df['column 1'] + df['column 2'] ) : column 1 값과 column 2 값을 각 더해 New column 이라는 새로운 파생변수 생성
• regexp_extract(col('column 1'), '정규표현식 패턴') : column 1이라는 칼럼 값에서 지정해준 정규표현식 패턴에 해당하는 값들만 추출(extract)해라.(여기서 col('column 1') 을 df['column 1'] 로 표현해도 동일한 표현이다.)

 

여기서는 탑승객들의 이름 중 Mr, Mrs와 같은 성별과 연령 의미를 내포하고 있는 salutation만을 추출하자.(salutation은 우리말로 '인사말'이며 여기서는 사람 이름의 가장 맨 앞에 나오는 Mr, Mrs 같은 것들을 의미한다.)

 

# 새로운 파생변수를 생성한 데이터프레임을 새로 할당!
df = df.withColumn("Initial", regexp_extract(col("Name"),
                                            "([A-Za-z]+)\.", # ()이게 하나의 그룹임!
                                            1)) # 그룹 인덱스는 1부터!
df.limit(3).show()

 

Initial 파생변수 만든 결과

 

추가적으로 탑승객들의 salutation 값에 오탈자가 있어 replace([오탈자], [수정된 글자]) 메소드를 통해 수정해준다. 수정할 문자열과 바로잡은 문자열을 리스트 자료구조로 여러개 부여할 수 있다.

 

df = df.replace(['Mlle','Mme', 'Ms', 'Dr','Major','Lady','Countess','Jonkheer','Col','Rev','Capt','Sir','Don'],
                ['Miss','Miss','Miss','Mr','Mr',  'Mrs',  'Mrs',  'Other',  'Other','Other','Mr','Mr','Mr'])

# Initial 변수 값들로 그룹핑한 후 평균 Age 구하기
df.groupby('Initial').avg('Age').collect()

 

collect() 의 결과 화면

 

위 코드에서 collect() 가 등장한다. collect() 도 select() 와 마찬가지로 일부의 데이터를 추출하는 역할을 한다. 하지만 사용목적에 따라 약간의 차이점이 존재한다. collect() 메소드는 다음과 같은 경우에 사용하는 것이 권장된다.

 

• 적은 양의 데이터셋을 반환할 때 자주 사용. 큰 데이터셋을 로드할 때는 메모리 에러가 발생할 가능성이 높다.
• 보통 filter(), group(), count() 메소드를 사용한 후 같이 자주 사용된다.
• 인자에 아무것도 넣지 않으면 즉, collect() 자체로만 사용한다면 해당 데이터프레임의 모든 row를 반환한다.
• collect() 가 반환하는 타입은 Spark의 데이터프레임 형태가 아닌 PySpark의 Row 타입의 데이터가 담긴 list를 반환한다. list 자료구조이기 때문에 loop 문으로 활용이 가능하다. 또한 PySpark의 Row 타입은 Python의 named Tuple 형태이다.

위 코드에서 탑승객 이름의 salutation 별로 Age 평균값을 계산했고 이를 이용해 결측치를 대체해보자.

 

df = df.withColumn('Age',
                  when((df['Initial'] == 'Miss') & (df['Age'].isNull()),
                      22).otherwise(df['Age']))
df = df.withColumn('Age',
                  when((df['Initial'] == 'Other') & (df['Age'].isNull()),
                      46).otherwise(df['Age']))
df = df.withColumn('Age',
                  when((df['Initial'] == 'Master') & (df['Age'].isNull()),
                      5).otherwise(df['Age']))
df = df.withColumn('Age',
                  when((df['Initial'] == 'Mr') & (df['Age'].isNull()),
                      33).otherwise(df['Age']))
df = df.withColumn('Age',
                  when((df['Initial'] == 'Mrs') & (df['Age'].isNull()),
                      36).otherwise(df['Age']))

 

결측치가 존재하는 칼럼이 Age 이외에 Embarked 에도 있었는데 Embarked의 결측치를 한 번 살펴보자.

 

# Embarked 변수에도 결측치가 2개 있었는데 무엇인지 확인해보기
df.groupBy('Embarked').count().show()

 

Embarked의 결측치는 2개이다.

 

결측치가 2개밖에 되지 않기 때문에 결측치는 Embarked의 최빈값인 'S' 값으로 대체해주자. 결측치를 대체하는 또 다른 메소드는 다음과 같다. df.na.fill({'column' : 'value'})

 

# Embarked의 결측치는 최빈값인 'S'로 대체해주기
df = df.na.fill({"Embarked": "S"})
# 결측치가 채워졌는지 다시 확인
df.groupBy('Embarked').count().show()

 

결측치를 잘 대체했다!

4. Feature Engineering 하기

위 3번이랑 일부 중복되는 내용일 수도 있다. 3번 목차에서 Name 칼럼을 이용해 Initial 이라는 새로운 파생변수를 생성한 것처럼 withcolumn, when, otherwise 를 사용해 Feature Engineering 을 수행해줄 수 있다.

 

# Family size라는 파생변수 생성
df = df.withColumn("Family_Size",
                  col('SibSp')+col('Parch')) # df['SibSp']도 가능!

# Alone이라는 Binary 파생변수 생성하는데, 우선 0으로 다 해놓기
df = df.withColumn('Alone', lit(0))
# 조건에 맞게 Alone 변수값 변경
df = df.withColumn('Alone',
                  when(col('Family_Size') == 0, 1)\
                  .otherwise(col('Alone')))

 

다음은 문자열로 되어있는 변수인 Sex, Embarked, Initial 값을 숫자로 변환해주는 즉, Label Encoding을 해주는 방법에 대해 알아보자. StringIndexer() 메소드를 사용하면 된다. 이름 그대로 String을 Index(숫자)로 변환해주는 기능을 제공한다. 자세한 사용법은 하단의 코드를 참고하자.

 

convert_cols = ['Sex', 'Embarked', 'Initial']

# 추후에 IndexToString할려면 indexer 객체를 사용하면 됨! 
indexer = [StringIndexer(inputCol=col,
                         outputCol=col+'_index').fit(df) for col in convert_cols]
for i in indexer:
    print(i)
    print('-'*80)

print(type(indexer))

 

칼럼 별로 StringIndexer가 만들어진다.

 

위 결과값 화면에서 uid를 보면 세 개가 모두 각기 다른 값을 갖는 것을 알 수 있다. 즉, 칼럼 별로 고유한 StringIndexer가 만들어졌음을 알 수 있다. 이제 fit() 과 transform() 메소드를 이용해 Label Encoding 을 수행해보자.

 

# Pipeline을 이용해 stage에다가 실행 과정 담아 넘기기
pipeline = Pipeline(stages=indexer)
df = pipeline.fit(df).transform(df)

5. 불필요한 칼럼들 삭제 후 최종 Feature들을 Vector로 변환하기

PySpark에서 여러개의 칼럼을 삭제하려면 Pandas와는 약간 다른 방법을 사용해야 편리하다. 우선 2가지 방법의 차이점을 알아보자.

 

• Pandas : df.drop(['columnA', 'columnB', 'columnC'], axis=1)
• PySpark : df.drop('columnA', 'columnB', 'columnC')

 

즉, drop() 메소드 안에 [](리스트)가 들어가는지 여부의 차이다. PySpark는 여러개의 칼럼을 삭제할 때도 [](리스트)로 감싸주어선 안된다. 따라서 Native Python의 기능 중 하나인 *(unpacking) 을 사용하면 된다.

 

unpacking을 사용하는 이유는 물론 불필요한 칼럼이 한, 두개 정도 일 때는 직접 정의해주어도 되지만 만약 100개, 200개로 늘어난다면 불필요한 칼럼들만 리스트로 추출한 후 unpacking을 수행하면 일일이 입력하는 시간을 줄일 수 있을 것이다.

 

un_cols = ["PassengerId","Name","Ticket","Cabin","Embarked","Sex","Initial"]

df = df.drop(*un_cols)
print("삭제 후 남은 칼럼들:", df.columns)

 

이제 불필요한 변수들도 삭제했고 최종 남은 Feature들을 Vector로 변환시켜 머신러닝 모델에 입력시킬 준비를 해보자. VectorAssembler() 메소드를 사용하면 되는데, 파라미터 인자가 StringIndexer()랑 비슷하지만 세부적으로 다음과 같은 2가지 차이점이 존재한다.

 

1. 파라미터 인자 중 inputCol 끝에 s가 붙은 inputCols 이다.
2. VectorAssembler를 정의해주고 수행해 줄 때 fit을 하지 않고 바로 transform을 수행해준다.

feature = VectorAssembler(inputCols = df.columns[1:],
                         outputCol='features')
feature_vector = feature.transform(df) # 데이터프레임 형태로 반환
print('feature type:', type(feature))
print('feature_vector type', type(feature_vector))

feature_vector.limit(3).toPandas()

 

Feature들을 Vector로 변환하기

 

위 결과의 빨간색 네모칸을 보게 되면 모든 Feature들에 대해 하나의 벡터로 만든 것을 볼 수 있다. 그런데 결과값을 확인하다가 특이한 점을 발견했다. features 값들 중 어떤 것들은 Tuple로 감싸진 벡터가, 어떤 것들은 List로 감싸진 벡터가 들어있었다. 구글링을 해봐도 모르기에 StackOverflow에 질문을 올려놓고 답변을 현재 기다리고 있다..(혹시 아시는 분 있다면 댓글로 답변해주시면 너무나도 감사하겠습니다!)

6. Train, Test 데이터 분할

이제 Feature들을 벡터화 시켰고 데이터를 학습, 테스트용으로 분할해보자. randomSplit([train_ratio, test_ratio]) 메소드를 활용하면 된다.

 

titanic_df = feature_vector.select(['features', 'Survived'])

# split train, test
(train_df, test_df) = titanic_df.randomSplit([0.8, 0.2], seed=42)

7. 머신러닝 분류 모델 만들기

이제 이진 분류 모델을 만들어보자. PySpark에서는 ParamGridBuilder() 를 통해 하이퍼파라미터 튜닝을 수행하면서 TrainValidationSplit() 메소드로 교차검증을 동시에 수행할 수 있다.

 

# 분류 모델 
from pyspark.ml.classification import LogisticRegression
# 파라미터 튜닝 & 교차 검증
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit
from pyspark.ml.tuning import CrossValidator
# 파이프라인
from pyspark.ml import Pipeline
# 메트릭 얻는 라이브러리
from pyspark.ml.evaluation import BinaryClassificationEvaluator
# ROC AUC(Sklearn)
from sklearn.metrics import roc_curve, auc


# 모델 정의
lr = LogisticRegression(labelCol='Survived')

# 튜닝할 파라미터 grid 정의
                                    # model.parameter 식으로 정의
paramGrid = ParamGridBuilder().addGrid(lr.regParam,
                                      (0.01, 0.1))\
                              .addGrid(lr.maxIter,
                                      (5, 10))\
                              .addGrid(lr.tol,
                                      (1e-4, 1e-5))\
                              .addGrid(lr.elasticNetParam,
                                      (0.25, 0.75))\
                              .build()

# 교차검증 정의 - Pipeline식으로 정의
tvs = TrainValidationSplit(estimator=lr,
                          estimatorParamMaps=paramGrid,
                          evaluator=MulticlassClassificationEvaluator(labelCol='Survived'),
                          trainRatio=0.8)
# 학습은 fit으로!
model = tvs.fit(train_df)
# 평가는 transform으로!
model_prediction = model.transform(test_df)

# 메트릭 평가
print('Accuracy:',
     MulticlassClassificationEvaluator(labelCol='Survived',
                                      metricName='accuracy').evaluate(model_prediction))
print('Precision:',
     MulticlassClassificationEvaluator(labelCol='Survived',
                                      metricName='weightedPrecision').evaluate(model_prediction))

 

Logistic Regression의 분류 성능

 

필자는 또 다른 분류 메트릭으로 AUC Score를 계산하기 위해 데이터가 Positive(1)로 예측될 확률값이 필요했다. 그래서 위 코드 중 model_prediction 객체에 할당된 값들을 로드해 보았다.

 

model_prediction.show(10)

 

model_prediction 변수에 반환된 값들

 

위 결과값 데이터프레임에서 변수가 의미하는 바는 다음과 같다.

 

• featrues : (10, [0,1,2,4,5], ... ) : 10개의 feature가 존재하고 각 feature의 값들은 0,1,2,4,5 이다. 이에 대한 설명은 StackOverflow를 참조해보자.
• rawPrediction : 해당 feature를 회귀 모델에 넣었을 때 계산되어 나오는 Raw한 결과값
• probability : rawPrediction 값에 로지스틱 함수를 적용한 후 변한된 값. 즉 0과 1사이의 확률값으로 매핑된 값
• prediction : probability가 특정 임계값 기준에 의해 1 또는 0으로 분류된 클래스(label)

8. ROC Curve 시각화하고 AUC Score 계산하기

이제 위 테이블에서 probability 와 Survived 칼럼들만 뽑아서 AUC 계산에 필요한 label 값과 Positive(1)로 예측될 확률값만을 뽑아보자.

 

# Evaluate ROC metric
from pyspark.mllib.evaluation import BinaryClassificationMetrics as metric
from pyspark import SparkContext

# SparkContext를 만들기
sc = SparkContext.getOrCreate()

# ROC 점수인 AUC를 계산 위해 Logistic를 적용해 나온 확률값과 레이블만 가져오기
results = model_prediction.select(['probability', 'Survived'])

# 확률값 - 레이블 set 준비시키기
# collect()로 모든 데이터 row retrieve(반환) - 리스트 형태로 반환
results_collect = results.collect()

# named tuple 형식이기 때문에 key 값으로 원하는 값을 불러올 수 있다!
print(results_collect[0])
print()
print('probability:', results_collect[0].probability)
print('Survived:', results_collect[0].Survived)

 

결과화면

 

위 결과화면에서 파란색 네모칸은 Negative(0)으로 예측 될 확률, 빨간색 네모칸은 Positive(1)로 예측될 확률을 의미한다. 따라서 AUC 계산을 위해 필요한 부분은 빨간색 네모칸과 Survived 값, 이 2가지다. 이제 필요한 값이 어디에 위치한지 알게 되었으니 이를 이용해 ROC Curve 와 AUC 계산을 해보자.

 

# Evaluate ROC metric
from pyspark.mllib.evaluation import BinaryClassificationMetrics as metric
from pyspark import SparkContext

# SparkContext를 만들기
sc = SparkContext.getOrCreate()

# ROC 점수인 AUC를 계산 위해 Logistic를 적용해 나온 확률값과 레이블만 가져오기
results = model_prediction.select(['probability', 'Survived'])

# 확률값 - 레이블 set 준비시키기
# collect()로 모든 데이터 row retrieve(반환) - 리스트 형태로 반환
results_collect = results.collect()
results_list = [(float(i.probability[1]),
                 float(i.Survived)) for i in results_collect]
# 여러개의 튜플이 담긴 list를 RDD 자료구조로 변경
scoreAndLabels = sc.parallelize(results_list)
# ROC metric 계산하기
metrics = metric(scoreAndLabels)
auc = metrics.areaUnderROC

# Visualize ROC Curve
from sklearn.metrics import roc_curve, auc

# roc_curve 는 실제값, 1로의 예측확률값을 인자로 넣어주면 FPR, TPR, 임곗값을 반환해줌
fpr = []
tpr = []
roc_auc = []

y_test = [i[1] for i in results_list]
y_proba = [i[0] for i in results_list]

fpr, tpr, _ = roc_curve(y_test, y_proba)
roc_auc = auc(fpr, tpr)

plt.figure()
# x축엔 Fall-out(FPR), y축엔 Recall(TPR)
plt.plot(fpr, tpr, label='ROC Curve (area = %0.2f)' % roc_auc)
plt.plot([0, 1], [0, 1], 'k--')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title("Area under the ROC Curve")
plt.legend(loc='lower right')
plt.show()

 

ROC Curve 와 AUC Score 계산

 

이렇게 Titanic 데이터를 활용한 머신러닝 분류 모델을 PySpark를 활용해 만들고 성능평가까지 진행해보았다. 해당 포스팅에 사용된 Logistic Regression 이외에 추가적으로 사용해본 Random Forest, XGBoost 모델은 Kaggle 노트북 원본이나 필자의 Github 코드를 살펴보면 도움이 될 것이다. 기본적인 모델 빌딩, 평가 프레임은 동일하기 때문에 모델 파라미터만 변경하면 된다.