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
.option("header", "true") // Use first line of all files as header
.option("inferSchema", "true") // Automatically infer data types

  // 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"))

  val labelIndexer2 = new StringIndexer()

  val labelIndexer1 = new StringIndexer()

  val rf = new RandomForestClassifier()

  val labelConverter = new IndexToString()

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

  // 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()
  val accuracy = evaluator.evaluate(predictions)
  println("Accuracy  = " + (accuracy))
  println("Test Error = " + (1.0 - accuracy))


  • \$\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.