5
\$\begingroup\$

Below is the code I have for a RandomForest multiclass-classification model. I am reading from a CSV file and doing various transformations as seen in the code. I am planning to run this in a Spark cluster with lots of data (at least 6 million rows of data with 50 features). I can get sufficient amount of RAM and cores so that is not a problem.

What parts of my code can be improved for scalability/efficiency in terms of spark?

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

  val conf = new SparkConf().setAppName("MLApp").setMaster("local")
  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)
  println(s"Running Spark Version ${sc.version}")

  // Need to predict 2 classes
  val cols_to_predict=Array("Label1","Label2")

  // ID col
  val omit_cols=Array("Key")

  // reading the csv file
  val data = sqlContext.read
.format("com.databricks.spark.csv")
.option("header", "true") // Use first line of all files as header
.option("inferSchema", "true") // Automatically infer data types
.load("abc.csv")
    .cache()
  println(data.printSchema())

  // creating a features DF by droppping the labels so that I can run all 
  // the cols through String Indexer
  val features=data.drop("Label1").drop("Label2").drop("Key")

  // Since I do not know my max categories possible, I find it out
  // and use it for maxBins parameter in RF
  val distinct_col_counts=features.columns.map(x => data.select(x).distinct().count ).max

  val transformers: Array[org.apache.spark.ml.PipelineStage] = features.columns.map(
    cname => new StringIndexer().setInputCol(cname).setOutputCol(s"${cname}_index").fit(features)
  )
  val assembler  = new VectorAssembler()
    .setInputCols(features.columns.map(cname => s"${cname}_index"))
    .setOutputCol("features")

  val labelIndexer2 = new StringIndexer()
    .setInputCol("prog_label2")
    .setOutputCol("Label2")
    .fit(data)

  val labelIndexer1 = new StringIndexer()
    .setInputCol("orig_label1")
    .setOutputCol("Label1")
    .fit(data)

  val rf = new RandomForestClassifier()
    .setLabelCol("Label1")
    .setFeaturesCol("features")
    .setNumTrees(100)
    .setMaxBins(distinct_col_counts.toInt)

  val labelConverter = new IndexToString()
    .setInputCol("prediction")
    .setOutputCol("predictedLabel")
    .setLabels(labelIndexer1.labels)

  // Split into train and test
  val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))
  trainingData.cache()
  testData.cache()

  // Running only for one label for now Label1
  val stages: Array[org.apache.spark.ml.PipelineStage] =transformers :+ labelIndexer1 :+ assembler :+ rf :+ labelConverter //:+ labelIndexer2

  val pipeline=new Pipeline().setStages(stages)
  val model=pipeline.fit(trainingData)
  val predictions = model.transform(testData)

  val evaluator = new MulticlassClassificationEvaluator()
    .setLabelCol("Label1")
    .setPredictionCol("prediction")
    .setMetricName("precision")
  val accuracy = evaluator.evaluate(predictions)
  println("Accuracy  = " + (accuracy))
  println("Test Error = " + (1.0 - accuracy))






  }

}
\$\endgroup\$
  • \$\begingroup\$ Do you have only categorical features? \$\endgroup\$ – zero323 Dec 3 '15 at 19:56
  • \$\begingroup\$ @Hugs: Your accounts are pending merging. Please wait a while before attempting to comment. \$\endgroup\$ – Jamal Dec 3 '15 at 20:46
  • \$\begingroup\$ @zero323 Yes, I have only Categorical features \$\endgroup\$ – Huga Dec 5 '15 at 19:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.