// ---
// jupyter:
//   jupytext:
//     text_representation:
//       extension: .scala
//       format_name: light
//       format_version: '1.5'
//       jupytext_version: 1.14.4
//   kernelspec:
//     display_name: Scala 2.11 with Spark 2.1
//     language: scala
//     name: scala-spark21
// ---

def setHadoopConfigd3bd5b94a9334de59a55a7fed2bedeaa(name: String) = {
    // This function sets the Hadoop configuration so it is possible to
    // access data from Bluemix Object Storage using Spark

    val prefix = "fs.swift.service." + name
    sc.hadoopConfiguration.set(prefix + ".auth.url", "https://identity.open.softlayer.com" + "/v3/auth/tokens")
    sc.hadoopConfiguration.set(prefix + ".auth.endpoint.prefix","endpoints")
    sc.hadoopConfiguration.set(prefix + ".tenant", "6aaf54352357483486ee2d4981f8ef15")
    sc.hadoopConfiguration.set(prefix + ".username", "9fea4ce0afb1415aac1a9c2dffb84be4")
    sc.hadoopConfiguration.set(prefix + ".password", "<PASSWORD>")
    sc.hadoopConfiguration.setInt(prefix + ".http.port", 8080)
    sc.hadoopConfiguration.set(prefix + ".region", "dallas")
    sc.hadoopConfiguration.setBoolean(prefix + ".public", false)
}
val name = "keystone"
setHadoopConfigd3bd5b94a9334de59a55a7fed2bedeaa(name)
var df_categorical = spark.read.parquet("swift://sparkbuch." + name + "/train_categorical2.0.parquet")
var df_numeric = spark.read.parquet("swift://sparkbuch." + name + "/train_numeric2.0.parquet")

df_categorical.printSchema
df_numeric.printSchema

df_numeric.first

df_categorical.createOrReplaceTempView("dfcat")
var dfcat = spark.sql("select Id, L0_S22_F545 from dfcat")

df_numeric.createOrReplaceTempView("dfnum")
var dfnum = spark.sql("select Id,L0_S0_F0,L0_S0_F2,L0_S0_F4,Response from dfnum")


// +
var df = dfcat.join(dfnum,"Id")
df.createOrReplaceTempView("df")

var df_notnull = spark.sql("""
select
    Response as label,
    case 
       when L0_S22_F545 is null then 'NA'
       when L0_S22_F545 = '' then 'NA' 
       else L0_S22_F545 end as L0_S22_F545, 
    case
       when L0_S0_F0 is null then 0.0 
       else L0_S0_F0 end as L0_S0_F0, 
    case
       when L0_S0_F2 is null then 0.0 
       else L0_S0_F2 end as L0_S0_F2,
    case
       when L0_S0_F4 is null then 0.0 
       else L0_S0_F4 end as L0_S0_F4
from df
""")

// +
import org.apache.spark.ml.feature.{OneHotEncoder, StringIndexer}

val indexer = new StringIndexer(). 
  setInputCol("L0_S22_F545"). 
  setOutputCol("L0_S22_F545Index")

val indexed = indexer.setHandleInvalid("skip").fit(df_notnull).transform(df_notnull)

// -

indexed.printSchema

indexed.first

indexed.select("L0_S22_F545Index").show

var encoder = new OneHotEncoder().setInputCol("L0_S22_F545Index").setOutputCol("L0_S22_F545Vec")
val encoded = encoder.transform(indexed)

encoded.printSchema

encoded.first

// +
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg.Vectors

val transformers = Array(
    indexer,
    encoder,
    new VectorAssembler()
        .setInputCols(Array("L0_S22_F545Vec", "L0_S0_F0", "L0_S0_F2","L0_S0_F4"))
        .setOutputCol("features")
)

// +
var pipeline = new Pipeline().setStages(transformers).fit(df_notnull)

var transformed = pipeline.transform(df_notnull)
// -

transformed.printSchema

transformed.first

// +
import org.apache.spark.ml.classification.RandomForestClassifier
val rf = new RandomForestClassifier(). 
  setLabelCol("label").
  setFeaturesCol("features")

val model = new Pipeline().setStages(transformers :+ rf).fit(df_notnull)

val result = model.transform(df_notnull)

// +
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
val evaluator = new BinaryClassificationEvaluator()

import org.apache.spark.ml.param.ParamMap
var evaluatorParamMap = ParamMap(evaluator.metricName -> "areaUnderROC")
var aucTraining = evaluator.evaluate(result, evaluatorParamMap)
// -

print(aucTraining)

// +
import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
/**var paramGrid = new ParamGridBuilder()
    .addGrid(rf.numTrees, 3 :: 5 :: 10 :: 30 :: 50 :: 70 :: 100 :: 150 :: Nil)
    .addGrid(rf.featureSubsetStrategy, "auto" :: "all" :: "sqrt" :: "log2" :: "onethird" :: Nil)
    .addGrid(rf.impurity, "gini" :: "entropy" :: Nil)    
    .addGrid(rf.maxBins, 2 :: 5 :: 10 :: 15 :: 20 :: 25 :: 30 :: Nil)
    .addGrid(rf.maxDepth, 3 :: 5 :: 10 :: 15 :: 20 :: 25 :: 30 :: Nil)
    .build()*/

var paramGrid = new ParamGridBuilder().
    addGrid(rf.numTrees, 3 :: 5 :: 10 :: Nil).
    addGrid(rf.featureSubsetStrategy, "auto" :: "all" :: Nil).
    addGrid(rf.impurity, "gini" :: "entropy" :: Nil).    
    addGrid(rf.maxBins, 2 :: 5 :: Nil).
    addGrid(rf.maxDepth, 3 :: 5 :: Nil).
    build()
// -

var crossValidator = new CrossValidator().
      setEstimator(new Pipeline().setStages(transformers :+ rf)).
      setEstimatorParamMaps(paramGrid).
      setNumFolds(5).
      setEvaluator(evaluator)

//Model is created
    var crossValidatorModel = crossValidator.fit(df_notnull)
    //Model used to Predict
    var newPredictions = crossValidatorModel.transform(df_notnull)


    var newAucTest = evaluator.evaluate(newPredictions, evaluatorParamMap)
    println("new AUC (with Cross Validation) " + newAucTest)
    var bestModel = crossValidatorModel.bestModel

    //Understand the Model selected
    println()
    println("Parameters for Best Model:")

    var bestPipelineModel = crossValidatorModel.bestModel.asInstanceOf[PipelineModel]
    var stages = bestPipelineModel.stages

import org.apache.spark.ml.classification.RandomForestClassificationModel
    val rfStage = stages(stages.length-1).asInstanceOf[RandomForestClassificationModel]
rfStage.getNumTrees
rfStage.getFeatureSubsetStrategy
rfStage.getImpurity
rfStage.getMaxBins
rfStage.getMaxDepth



StarcoderdataJupyterScriptsDedupFiltered