1. Spark Session Build
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Titanic Binary Classification example") \
.getOrCreate()
2. Spark read csv
data = spark.read.csv('./data/titanic.csv', header=True, inferSchema=True)
3. 데이터프레임 검정통계량
data.select(['*']).describe().show()
4. 데이터 클린업
- PassengerID, Name, Ticket, Embarked는 사용하지 않을 예정 (아무 의미가 없음).
- Cabin도 비어있는 값이 너무 많아서 사용하지 않을 예정
- Age는 중요한 정보인데 비어있는 레코드들이 많아서 디폴트값을 채워줄 예정
- Gender의 경우 카테고리 정보이기에 숫자로 인코딩 필요
(1) 사용 데이터프레임 조정
final_data = data.select(['Survived', 'Pclass', 'Gender', 'Age', 'SibSp', 'Parch', 'Fare'])
(2) 특정 컬럼 null 값 채우기
from pyspark.ml.feature import Imputer
# MISSING VALUE를 채우는 imputer
# 평균값으로, Age 컬럼을 채울거고, 새로운 컬럼을 붙이는데 이름은 AgeImputed다
# imputer 인스턴스를 만들고, imputer.fit
# transform
imputer = Imputer(strategy='mean', inputCols=['Age'], outputCols=['AgeImputed'])
imputer_model = imputer.fit(final_data)
final_data = imputer_model.transform(final_data)
- null 값 채우는데에는 Imputer 메소드를 활용한다.
(3) 성별 정보 인코딩
# string을 indexer로 만드는 것
from pyspark.ml.feature import StringIndexer
gender_indexer = StringIndexer(inputCol='Gender', outputCol='GenderIndexed')
gender_indexer_model = gender_indexer.fit(final_data)
final_data = gender_indexer_model.transform(final_data)
5. 모델 학습
(1) 피처 벡터 생성
from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=['Pclass', 'SibSp', 'Parch', 'Fare', 'AgeImputed', 'GenderIndexed'], outputCol='features')
data_vec = assembler.transform(final_data)
(2) 훈련/테스트용 데이터 분리, 모델 적용
train, test = data_vec.randomSplit([0.7, 0.3])
from pyspark.ml.classification import LogisticRegression
algo = LogisticRegression(featuresCol="features", labelCol="Survived")
model = algo.fit(train)
(3) 모델 성능 측정
predictions = model.transform(test)
from pyspark.ml.evaluation import BinaryClassificationEvaluator
# ROC 그래프를 그려보고
# AUC를 구한다
evaluator = BinaryClassificationEvaluator(labelCol='Survived', metricName='areaUnderROC')
evaluator.evaluate(predictions)
