Spark ML 04(Pyspark)

Spark ML을 이용하여 Decision Tree를 수행하는 예제입니다.

---

Decision Tree(의사결정나무)

  결정 트리 학습법은 어떤 항목에 대한 관측값과 목표값을 연결시켜주는 예측 모델로써 결정 트리를 사용한다. 이는 통계학과 데이터 마이닝, 기계 학습에서 사용하는 예측 모델링 방법 중 하나이다. 트리 모델 중 목표 변수가 유한한 수의 값을 가지는 것을 분류 트리라 한다. 이 트리 구조에서 잎(리프 노드)은 클래스 라벨을 나타내고 가지는 클래스 라벨과 관련있는 특징들의 논리곱을 나타낸다. 결정 트리 중 목표 변수가 연속하는 값, 일반적으로 실수를 가지는 것은 회귀 트리라 한다.

  의사 결정 분석에서 결정 트리는 시각적이고 명시적인 방법으로 의사 결정 과정과 결정된 의사를 보여주는데 사용된다. 데이터 마이닝 분야에서 결정 트리는 결정된 의사보다는 자료 자체를 표현하는데 사용된다. 다만, 데이터 마이닝의 결과로서의 분류 트리는 의사 결정 분석의 입력 값으로 사용될 수 있다. 이 페이지는 데이터 마이닝 분야에서의 결정 트리를 주로 다룬다.

타이타닉 호 탑승객의 생존 여부를 나타내는 결정 트리. (“sibsp”는 탑승한 배우자와 자녀의 수를 의미한다.) 잎 아래의 숫자는 각각 생존 확률과 탑승객이 그 잎에 해당될 확률을 의미한다.

https://ko.wikipedia.org/wiki/%EA%B2%B0%EC%A0%95_%ED%8A%B8%EB%A6%AC_%ED%95%99%EC%8A%B5%EB%B2%95 

결정 트리 학습법 - 위키백과, 우리 모두의 백과사전

Spark ML 예제

 

 

In [1]:

cuse = spark.read.csv("data/cuse_binary.csv", inferSchema=True, header=True)

In [3]:

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml import Pipeline

In [4]:

categorical_columns = cuse.columns[0:3]

In [6]:

stringindexer_stages = [StringIndexer(inputCol=c, \
                                      outputCol='stringindexed_'+c) \
                        for c in categorical_columns]
stringindexer_stages += [StringIndexer(inputCol='y', outputCol='label')]

In [7]:

onehotencoder_stages = [OneHotEncoder(inputCol='stringindexed_'+c, \
                                      outputCol='onehot_'+c)
                        for c in categorical_columns]

In [9]:

feature_columns = ['onehot_'+c for c in categorical_columns]

In [10]:

vectorassembler_stage = VectorAssembler(inputCols=feature_columns, outputCol='features')

In [11]:

all_stages = stringindexer_stages + onehotencoder_stages + [vectorassembler_stage]

In [12]:

pipeline = Pipeline(stages=all_stages)

In [13]:

pipeline_model = pipeline.fit(cuse)

In [14]:

final_columns = feature_columns + ['features', 'label']

In [15]:

cuse_df = pipeline_model.transform(cuse).select(final_columns)

In [16]:

trainDF, testDF = cuse_df.randomSplit([0.8, 0.2])

In [17]:

from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.classification import 

In [20]:

dt = DecisionTreeClassifier(featuresCol="features", labelCol='label')

In [19]:

from pyspark.ml.tuning import ParamGridBuilder

In [21]:

param_grid = ParamGridBuilder() \
            .addGrid(dt.maxDepth, [2, 3, 4, 5]).build()

In [22]:

from pyspark.ml.evaluation import BinaryClassificationEvaluator

In [27]:

evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", \
                                          metricName="areaUnderROC")

In [28]:

from pyspark.ml.tuning import CrossValidator

In [29]:

cv = CrossValidator(estimator=dt, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)

In [30]:

cv_model = cv.fit(cuse_df)

In [31]:

pred_train_df = cv_model.transform(trainDF)

In [33]:

pred_train_df.select(['features', 'label', 'prediction', 'rawPrediction']).show(5)

 

+---------+-----+----------+-------------+
| features|label|prediction|rawPrediction|
+---------+-----+----------+-------------+
|(5,[],[])|  0.0|       1.0|[203.0,237.0]|
|(5,[],[])|  0.0|       1.0|[203.0,237.0]|
|(5,[],[])|  0.0|       1.0|[203.0,237.0]|
|(5,[],[])|  0.0|       1.0|[203.0,237.0]|
|(5,[],[])|  0.0|       1.0|[203.0,237.0]|
+---------+-----+----------+-------------+
only showing top 5 rows

In [34]:

cv_model.bestModel.featureImportances

Out[34]:

SparseVector(5, {1: 0.166, 2: 0.1888, 4: 0.6453})

In [ ]:

In [ ]:

In [ ]: