김용환 블로그(2004-2020) :: [spark] 스파크 머신 러닝(ML) api을 사용하여 파이프 라인 개발하기

[spark] 스파크 머신 러닝(ML) api을 사용하여 파이프 라인 개발하기 - 유방암 가능성 예측

scala 2018. 5. 31. 17:42

spark-shell을 이용해 유방암 가능성을 예측하는 방법을 소개한다.

https://archive.ics.uci.edu/ml/machine-learning-databases/breast-cancer-wisconsin/breast-cancer-wisconsin.data에서 데이터를 다운로드받고 로드한 다음 파싱하는 예를 소개한다.

breast-cancer-wisconsin.data를 읽고 이를 케이스 클래스로 저장하고 파싱하는 코드를 추가한다.

import org.apache.spark.rdd.RDD

case class Cancer(cancer_class: Double, thickness: Double, size: Double, shape: Double, madh: Double, epsize: Double, bnuc: Double, bchrom: Double, nNuc: Double, mit: Double)

def parseRDD(rdd: RDD[String]): RDD[Array[Double]] = {

rdd.map(_.split(",")).filter(_(6) != "?").map(_.drop(1)).map(_.map(_.toDouble))

}

def parseCancer(line: Array[Double]): Cancer = {

Cancer(if (line(9) == 4.0) 1 else 0, line(0), line(1), line(2), line(3), line(4), line(5), line(6), line(7), line(8))

}

val rdd = spark.sparkContext.textFile("data/breast-cancer-wisconsin.data")

val cancerRDD = parseRDD(rdd).map(parseCancer)

머신러닝 파이프 라인을 위해 RDD를 데이터 프레임으로 변환한다.

import spark.sqlContext.implicits._

val cancerDF = cancerRDD.toDF().cache()

cancerDF.show()

+------------+---------+----+-----+----+------+----+------+----+---+

+------------+---------+----+-----+----+------+----+------+----+---+

| 0.0| 5.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|

| 0.0| 5.0| 4.0| 4.0| 5.0| 7.0|10.0| 3.0| 2.0|1.0|

| 0.0| 3.0| 1.0| 1.0| 1.0| 2.0| 2.0| 3.0| 1.0|1.0|

| 0.0| 6.0| 8.0| 8.0| 1.0| 3.0| 4.0| 3.0| 7.0|1.0|

| 0.0| 4.0| 1.0| 1.0| 3.0| 2.0| 1.0| 3.0| 1.0|1.0|

| 1.0| 8.0|10.0| 10.0| 8.0| 7.0|10.0| 9.0| 7.0|1.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0|10.0| 3.0| 1.0|1.0|

| 0.0| 2.0| 1.0| 2.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 1.0| 1.0|5.0|

| 0.0| 4.0| 2.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 1.0| 1.0| 3.0| 1.0|1.0|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|

| 1.0| 5.0| 3.0| 3.0| 3.0| 2.0| 3.0| 4.0| 4.0|1.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 3.0| 3.0| 1.0|1.0|

| 1.0| 8.0| 7.0| 5.0|10.0| 7.0| 9.0| 5.0| 5.0|4.0|

| 1.0| 7.0| 4.0| 6.0| 4.0| 6.0| 1.0| 4.0| 3.0|1.0|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|

| 1.0| 10.0| 7.0| 7.0| 6.0| 4.0|10.0| 4.0| 1.0|2.0|

| 0.0| 6.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|

+------------+---------+----+-----+----+------+----+------+----+---+

only showing top 20 rows

피쳐를 추출하고 트랜스포메이션을 적용해보자.

먼저 다음처럼 피쳐 컬럼을 선택한다.

val featureCols = Array("thickness", "size", "shape", "madh", "epsize", "bnuc", "bchrom", "nNuc", "mit")

이제 다음처럼 피쳐 컬럼(featureCols)을 피쳐 벡터(features)로 생성한다.

import org.apache.spark.ml.feature.VectorAssembler

val assembler = new VectorAssembler().setInputCols(featureCols).setOutputCol("features")

이제 다음처럼 피쳐 벡터를 데이터 프레임으로 변환한다.

val df2 = assembler.transform(cancerDF)

df2.show()

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+

| 0.0| 5.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[5.0,1.0,1.0,1.0,...|

| 0.0| 5.0| 4.0| 4.0| 5.0| 7.0|10.0| 3.0| 2.0|1.0|[5.0,4.0,4.0,5.0,...|

| 0.0| 3.0| 1.0| 1.0| 1.0| 2.0| 2.0| 3.0| 1.0|1.0|[3.0,1.0,1.0,1.0,...|

| 0.0| 6.0| 8.0| 8.0| 1.0| 3.0| 4.0| 3.0| 7.0|1.0|[6.0,8.0,8.0,1.0,...|

| 0.0| 4.0| 1.0| 1.0| 3.0| 2.0| 1.0| 3.0| 1.0|1.0|[4.0,1.0,1.0,3.0,...|

| 1.0| 8.0|10.0| 10.0| 8.0| 7.0|10.0| 9.0| 7.0|1.0|[8.0,10.0,10.0,8....|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0|10.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...|

| 0.0| 2.0| 1.0| 2.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[2.0,1.0,2.0,1.0,...|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 1.0| 1.0|5.0|[2.0,1.0,1.0,1.0,...|

| 0.0| 4.0| 2.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[4.0,2.0,1.0,1.0,...|

| 0.0| 1.0| 1.0| 1.0| 1.0| 1.0| 1.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[2.0,1.0,1.0,1.0,...|

| 1.0| 5.0| 3.0| 3.0| 3.0| 2.0| 3.0| 4.0| 4.0|1.0|[5.0,3.0,3.0,3.0,...|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 3.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...|

| 1.0| 8.0| 7.0| 5.0|10.0| 7.0| 9.0| 5.0| 5.0|4.0|[8.0,7.0,5.0,10.0...|

| 1.0| 7.0| 4.0| 6.0| 4.0| 6.0| 1.0| 4.0| 3.0|1.0|[7.0,4.0,6.0,4.0,...|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[4.0,1.0,1.0,1.0,...|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[4.0,1.0,1.0,1.0,...|

| 1.0| 10.0| 7.0| 7.0| 6.0| 4.0|10.0| 4.0| 1.0|2.0|[10.0,7.0,7.0,6.0...|

| 0.0| 6.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[6.0,1.0,1.0,1.0,...|

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+

이제 왼쪽 컬럼을 기반으로 계산된 피쳐가 포함된 데이터 프레임을 관찰해야 한다.

마지막으로 StringIndexer를 사용하고 다음처럼 트레이닝 데이터셋 레이블을 생성한다.

import org.apache.spark.ml.feature.StringIndexer

val labelIndexer = new StringIndexer().setInputCol("cancer_class").setOutputCol("label")

val df3 = labelIndexer.fit(df2).transform(df2)

df3.show()

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+

| 0.0| 5.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[5.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 5.0| 4.0| 4.0| 5.0| 7.0|10.0| 3.0| 2.0|1.0|[5.0,4.0,4.0,5.0,...| 0.0|

| 0.0| 3.0| 1.0| 1.0| 1.0| 2.0| 2.0| 3.0| 1.0|1.0|[3.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 6.0| 8.0| 8.0| 1.0| 3.0| 4.0| 3.0| 7.0|1.0|[6.0,8.0,8.0,1.0,...| 0.0|

| 0.0| 4.0| 1.0| 1.0| 3.0| 2.0| 1.0| 3.0| 1.0|1.0|[4.0,1.0,1.0,3.0,...| 0.0|

| 1.0| 8.0|10.0| 10.0| 8.0| 7.0|10.0| 9.0| 7.0|1.0|[8.0,10.0,10.0,8....| 1.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0|10.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 2.0| 1.0| 2.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[2.0,1.0,2.0,1.0,...| 0.0|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 1.0| 1.0|5.0|[2.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 4.0| 2.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[4.0,2.0,1.0,1.0,...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 1.0| 1.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 2.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[2.0,1.0,1.0,1.0,...| 0.0|

| 1.0| 5.0| 3.0| 3.0| 3.0| 2.0| 3.0| 4.0| 4.0|1.0|[5.0,3.0,3.0,3.0,...| 1.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 3.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|

| 1.0| 8.0| 7.0| 5.0|10.0| 7.0| 9.0| 5.0| 5.0|4.0|[8.0,7.0,5.0,10.0...| 1.0|

| 1.0| 7.0| 4.0| 6.0| 4.0| 6.0| 1.0| 4.0| 3.0|1.0|[7.0,4.0,6.0,4.0,...| 1.0|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[4.0,1.0,1.0,1.0,...| 0.0|

| 0.0| 4.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[4.0,1.0,1.0,1.0,...| 0.0|

| 1.0| 10.0| 7.0| 7.0| 6.0| 4.0|10.0| 4.0| 1.0|2.0|[10.0,7.0,7.0,6.0...| 1.0|

| 0.0| 6.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[6.0,1.0,1.0,1.0,...| 0.0|

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+

이제 왼쪽 컬럼을 기반으로 계산된 피쳐와 레이블이 포함된 데이터 프레임을 살펴보자.

머신 러닝 모델을 트레이닝하기 위한 피쳐와 레이블이 포함된 새로운 데이터 프레임을 생성할 것이다.

먼저 테스트 셋과 트레이닝 셋(3:7)을 생성한다. (지도 학습이라..)

val splitSeed = 1234567

val Array(trainingData, testData) = df3.randomSplit(Array(0.7, 0.3), splitSeed)

이제 트레이닝 셋을 사용하여 에스티메이터 생성한다. elasticNetParam을 이용한 로지스틱 회귀 분석을 사용해 파이프 라인에 대한 에스티메이터를 생성하자. 다음처럼 최대 반복과 회귀 매개 변수도 지정한다.

import org.apache.spark.ml.classification.LogisticRegression

val lr = new LogisticRegression().setMaxIter(50).setRegParam(0.01).setElasticNetParam(0.01)

val model = lr.fit(trainingData)

테스트 셋으로 원본 예측, 확률, 예측을 얻기 위해 테스트 셋을 사용해 모델을 변환한다.

val predictions = model.transform(testData)

predictions.show()

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+--------------------+--------------------+----------+

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+--------------------+--------------------+----------+

| 0.0| 1.0| 1.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[5.15956430979038...|[0.99428860556932...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 1.0| 1.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.88229871718381...|[0.99247744702488...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0| 1.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[5.26929960916807...|[0.99487914377217...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.99203401656150...|[0.99325398211858...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0| 2.0| 3.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.74802132478210...|[0.99140567173413...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.71476842395493...|[0.99111766179519...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 1.0| 3.0| 2.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.59276207806523...|[0.98997663106901...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 2.0| 5.0| 1.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.10129026316119...|[0.98371817931939...| 0.0|

| 0.0| 1.0| 1.0| 1.0| 1.0| 4.0| 3.0| 1.0| 1.0|1.0|[1.0,1.0,1.0,1.0,...| 0.0|[4.35023434970686...|[0.98726059831436...| 0.0|

+------------+---------+----+-----+----+------+----+------+----+---+--------------------+-----+--------------------+--------------------+----------+

이제 원본 예측과 각 로우에 대한 실제 예측이 포함된 새로운 데이터 프레임을 볼 수 있다.

다음처럼 트레이닝의 객관적인 기록 생성을 위해 각 반복마다 모델의 목표 이력을 생성하자.

val trainingSummary = model.summary

val objectiveHistory = trainingSummary.objectiveHistory

objectiveHistory.foreach(loss => println(loss))

이전 코드는 트레이닝 손실 측면에서 다음과 같은 출력을 생성한다.

0.6562291876496595

0.6087867761081431

0.538972588904556

0.4928455913405332

0.46269258074999386

0.3527914819973198

0.20206901337404978

0.16459454874996993

0.13783437051276512

0.11478053164710095

0.11420433621438157

0.11138884788059378

0.11041889032338036

0.10849477236373875

0.10818880537879513

0.10682868640074723

0.10641395229253267

0.10555411704574749

0.10505186414044905

0.10470425580130915

0.10376219754747162

0.10331139609033112

0.10276173290225406

0.10245982201904923

0.10198833366394071

0.10168248313103552

0.10163242551955443

0.10162826209311404

0.10162119367292953

0.10161235376791203

0.1016114803209495

0.10161090505556039

0.1016107261254795

0.10161056082112738

0.10161050381332608

0.10161048515341387

0.10161043900301985

0.10161042057436288

0.10161040971267737

0.10161040846923354

0.10161040625542347

0.10161040595207525

0.10161040575664354

0.10161040565870835

0.10161040519559975

0.10161040489834573

0.10161040445215266

0.1016104043469577

0.1016104042793553

0.1016104042606048

0.10161040423579716

이전 결과를 살펴보면 반복하면거 결과 값이 점차 줄어드는 것을 볼 수 있다.

이제 이진 로직 회귀 요약에서 사용한 분류자인지 확인해야 한다.

import org.apache.spark.ml.classification.BinaryLogisticRegressionSummary

val binarySummary = trainingSummary.asInstanceOf[BinaryLogisticRegressionSummary]

이제 ROC를 데이터 프레임으로 얻고 areaUnderROC로 가져온다. 근사값이 1.0이면 더 좋다.

val roc = binarySummary.roc

roc.show()

+---+--------------------+

|FPR| TPR|

+---+--------------------+

|0.0| 0.0|

|0.0|0.017341040462427744|

|0.0| 0.03468208092485549|

|0.0| 0.05202312138728324|

|0.0| 0.06936416184971098|

|0.0| 0.08670520231213873|

|0.0| 0.10404624277456648|

|0.0| 0.12138728323699421|

|0.0| 0.13872832369942195|

|0.0| 0.15606936416184972|

|0.0| 0.17341040462427745|

|0.0| 0.1907514450867052|

|0.0| 0.20809248554913296|

|0.0| 0.2254335260115607|

|0.0| 0.24277456647398843|

|0.0| 0.26011560693641617|

|0.0| 0.2774566473988439|

|0.0| 0.2947976878612717|

|0.0| 0.31213872832369943|

|0.0| 0.32947976878612717|

+---+--------------------+

areaUnderROC의 값을 출력한다.

println("binarySummary.areaUnderROC)

0.9960056075125305

이제 참 긍정 비율, 거짓 긍정 비율, 거짓 부정 비율, 전체 개수와 같은 메트릭과 다음처럼 제대로 예측하고 잘못 예측한 경우의 수를 계산하자.

import org.apache.spark.sql.functions._

// 성능 메트릭을 계산한다

val lp = predictions.select("label", "prediction")

val counttotal = predictions.count()

val correct = lp.filter($"label" === $"prediction").count()

val wrong = lp.filter(not($"label" === $"prediction")).count()

val truep = lp.filter($"prediction" === 0.0).filter($"label" === $"prediction").count()

val falseN = lp.filter($"prediction" === 0.0).filter(not($"label" === $"prediction")).count()

val falseP = lp.filter($"prediction" === 1.0).filter(not($"label" === $"prediction")).count()

val ratioWrong = wrong.toDouble / counttotal.toDouble

val ratioCorrect = correct.toDouble / counttotal.toDouble

println("Total Count: " + counttotal)

println("Correctly Predicted: " + correct)

println("Wrongly Identified: " + wrong)

println("True Positive: " + truep)

println("False Negative: " + falseN)

println("False Positive: " + falseP)

println("ratioWrong: " + ratioWrong)

println("ratioCorrect: " + ratioCorrect)

결과는 다음과 같다.

Total Count: 209

Correctly Predicted: 202

Wrongly Identified: 7

True Positive: 140

False Negative: 4

False Positive: 3

ratioWrong: 0.03349282296650718

ratioCorrect: 0.9665071770334929

마지막으로 모델의 정확도를 판단하자. 그러나 먼저 fMeasure를 최대화하기 위해 모델 임계 값을 설정해야 한다.

val fMeasure = binarySummary.fMeasureByThreshold

val fm = fMeasure.col("F-Measure")

val maxFMeasure = fMeasure.select(max("F-Measure")).head().getDouble(0)

val bestThreshold = fMeasure.where($"F-Measure" === maxFMeasure).select("threshold").head().getDouble(0)

model.setThreshold(bestThreshold)

이제 다음처럼 정확도를 계산하자.

import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator

val evaluator = new BinaryClassificationEvaluator().setLabelCol("label")

val accuracy = evaluator.evaluate(predictions)

println("Accuracy: " + accuracy)

정확도는 다음처럼 거의 99.64%이다.

Accuracy: 0.9963975418520874

저작자표시 비영리 동일조건

'scala' 카테고리의 다른 글

[spark] 로지스틱 회귀 분석을 이용한 멀티 클래스 분류 (0)	2018.05.31
[spark] spark-shell 메모리/cpu 설정 (0)	2018.05.31
[spark] 머신러닝 - SGD(선형 회귀 기반 알고리즘) 적용 예 (0)	2018.05.30
[spark] 머신 러닝 - PCA 예 (0)	2018.05.30
[spark] 머신 러닝 - StringIndexer/One-host encoding 예 (0)	2018.05.29

Posted by '김용환'

[spark] 스파크 머신 러닝(ML) api을 사용하여 파이프 라인 개발하기 - 유방암 가능성 예측

'scala' 카테고리의 다른 글

카테고리

태그목록

최근에 올라온 글

최근에 달린 댓글

최근에 받은 트랙백

글 보관함

달력

링크

티스토리툴바