[Spark ML] Spark ML로 iris 붓꽃 데이터 예측 모델 만들기

import org.apache.spark.sql.{Column, SparkSession}
import org.apache.spark.sql.functions.{col, count, when}
import org.apache.spark.ml.classification._
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{StringIndexer, VectorAssembler}
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StringType, StructField, StructType}

object ml {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("SparkByExamples.com")
      .config("spark.driver.bindAddress", "127.0.0.1")
      .getOrCreate()

    val schema = StructType(Array(
      StructField("sepal_length", DoubleType, nullable = true),
      StructField("sepal_width", DoubleType, nullable = true),
      StructField("petal_length", DoubleType, nullable = true),
      StructField("petal_width", DoubleType, nullable = true),
      StructField("target", StringType, nullable = true)
    ))

    val df = spark.read
      .schema(schema)
      .csv("/Users/jang-won-yong/dev/ml/iris.data")
      .withColumn("target", when(col("target") === "Iris-setosa", "0")
        .when(col("target") === "Iris-virginica", "1")
        .otherwise("2")
      )
      .withColumn("target", col("target").cast(IntegerType))

    df.createOrReplaceTempView("df_view") // temp view 생성 하여 sql로 확인 가능
    spark.sql("select distinct target from df_view").show()

    df.show(truncate = false) // 데이터 확인

    df.printSchema() // 스키마 확인

    // randomSplit() 을 이용 하여 train 과 test 용 DataFrame 으로 분할
    val randomSplitArray = df.randomSplit(Array(0.8, 0.2), seed = 42)
    val trainDf = randomSplitArray(0) // 0.8
    val testDf = randomSplitArray(1)  // 0.2
    trainDf.cache()
    testDf.cache()

    // VectorAssembler 를 이용하여 모든 feature 컬럼들을 하나의 feature vector 로 변환
    val columns = Array("sepal_length", "sepal_width", "petal_length", "petal_width")
    val assembler = new VectorAssembler()
      .setInputCols(columns)
      .setOutputCol("features")

    val trainVectorDf = assembler.transform(trainDf)
    val testVectorDf = assembler.transform(testDf)

    // ML 알고리즘
    val dt = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setLabelCol("target")
      .setMaxDepth(5)

    val model = dt.fit(trainVectorDf)

    // 학습된 모델의 transform 메소드를 이용 하여 예측 수행
    val predictions = model.transform(testVectorDf)


    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("target")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")

    val accuracy = evaluator.evaluate(predictions)

    println("accuracy: " + accuracy)

    spark.close()
  }
}

val stage1 = new VectorAssembler()
  .setInputCols(columns)
  .setOutputCol("features")

// ML 알고리즘
val stage2 = new DecisionTreeClassifier()
  .setFeaturesCol("features")
  .setLabelCol("target")
  .setMaxDepth(5)

// Pipeline 은 여러개의 개별적인 Transformer 의 변환 작업, Estimator의 학습 작업을 일련의 Process로 연결을 통해
// 간단한 API 처리로 구현할 수 있게 만들어 줌
// Pipeline 은 개별 변환 및 학습 작업을 stage로 각각 정의하여 pipeline 에 등록한 뒤 pipeline fit() 메소드를 이용 호출하여
// 연결된 stage 작업을 수행
// 수행 결과로 PipelineModel 이 반환되며, 이 pipelineModel 에서 예측을 위한 변환 및 예측 작업을 transform() 메소드로 수행
val pipeline = new Pipeline()
  .setStages(Array(stage1, stage2)) // stage가 순차적으로 실행

val pipelineModel = pipeline.fit(trainDf)          
val predictions = pipelineModel.transform(testDf)  

// CrossValidator 에서 하이퍼 파라미터 튜닝을 위한 Grid Search 용 ParamGrid 생성
val paramGridMap = new ParamGridBuilder()
  .addGrid(dt.maxDepth, Array(5, 6, 7, 8, 9, 10)) // DecisionTreeClassifier 의 maxDepth
  .addGrid(dt.minInstancesPerNode, Array(3, 6)) // min samples split(노드 분할 시 최소 sample 건수)
  .build()

val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("target")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")

val stage2 = new CrossValidator()
  .setEstimator(dt)
  .setEstimatorParamMaps(paramGridMap)
  .setEvaluator(evaluator)
  .setNumFolds(3)

val pipeline = new Pipeline()
  .setStages(Array(stage1, stage2))

val pipelineModel = pipeline.fit(trainDf)
val predictions = pipelineModel.transform(testDf)

val accuracy = evaluator.evaluate(predictions)

println("accuracy: " + accuracy)

val cvModel = pipelineModel.stages(1).asInstanceOf[CrossValidatorModel]

printCVResult(cvModel)

def printCVResult(cvModel: CrossValidatorModel): Unit = {
    cvModel.getEstimatorParamMaps
      .zip(cvModel.avgMetrics)
      .sortBy(_._2)
      .foreach(println(_))}

import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder, TrainValidationSplit}

val trainValidationSplit = new TrainValidationSplit()
      .setEstimator(dt)
      .setEstimatorParamMaps(paramGridMap)
      .setEvaluator(evaluator)
      .setTrainRatio(0.75) // 학습 0.75, 검증 0.25
      .setSeed(0)