Cleanup Helloworld examples

helloworld/src/main/scala/com/salesforce/hw/boston/BostonFeatures.scala

@@ -34,35 +34,19 @@ import com.salesforce.op.features.FeatureBuilder
 import com.salesforce.op.features.types._
 
 trait BostonFeatures extends Serializable {
-
   val rowId = FeatureBuilder.Integral[BostonHouse].extract(_.rowId.toIntegral).asPredictor
-
   val crim = FeatureBuilder.RealNN[BostonHouse].extract(_.crim.toRealNN).asPredictor
-
   val zn = FeatureBuilder.RealNN[BostonHouse].extract(_.zn.toRealNN).asPredictor
-
   val indus = FeatureBuilder.RealNN[BostonHouse].extract(_.indus.toRealNN).asPredictor
-
   val chas = FeatureBuilder.PickList[BostonHouse].extract(x => Option(x.chas).toPickList).asPredictor
-
   val nox = FeatureBuilder.RealNN[BostonHouse].extract(_.nox.toRealNN).asPredictor
-
   val rm = FeatureBuilder.RealNN[BostonHouse].extract(_.rm.toRealNN).asPredictor
-
   val age = FeatureBuilder.RealNN[BostonHouse].extract(_.age.toRealNN).asPredictor
-
   val dis = FeatureBuilder.RealNN[BostonHouse].extract(_.dis.toRealNN).asPredictor
-
   val rad = FeatureBuilder.Integral[BostonHouse].extract(_.rad.toIntegral).asPredictor
-
   val tax = FeatureBuilder.RealNN[BostonHouse].extract(_.tax.toRealNN).asPredictor
-
   val ptratio = FeatureBuilder.RealNN[BostonHouse].extract(_.ptratio.toRealNN).asPredictor
-
   val b = FeatureBuilder.RealNN[BostonHouse].extract(_.b.toRealNN).asPredictor
-
   val lstat = FeatureBuilder.RealNN[BostonHouse].extract(_.lstat.toRealNN).asPredictor
-
   val medv = FeatureBuilder.RealNN[BostonHouse].extract(_.medv.toRealNN).asResponse
-
 }

helloworld/src/main/scala/com/salesforce/hw/boston/OpBoston.scala

@@ -36,7 +36,6 @@ import com.salesforce.op.readers.CustomReader
 import com.salesforce.op.stages.impl.regression.RegressionModelSelector
 import com.salesforce.op.stages.impl.regression.RegressionModelsToTry._
 import com.salesforce.op.stages.impl.tuning.DataSplitter
-import com.salesforce.op.utils.kryo.OpKryoRegistrator
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{Dataset, SparkSession}
 
@@ -45,52 +44,48 @@ import org.apache.spark.sql.{Dataset, SparkSession}
  */
 object OpBoston extends OpAppWithRunner with BostonFeatures {
 
-  override def kryoRegistrator: Class[_ <: OpKryoRegistrator] = classOf[BostonKryoRegistrator]
-
   ////////////////////////////////////////////////////////////////////////////////
   // READERS DEFINITION
   /////////////////////////////////////////////////////////////////////////////////
 
-  val randomSeed = 112233L
+  val randomSeed = 42L
 
-  def customRead(path: Option[String], spark: SparkSession): RDD[BostonHouse] = {
-    require(path.isDefined, "The path is not set")
-    val myFile = spark.sparkContext.textFile(path.get)
+  def customRead(path: String)(implicit spark: SparkSession): RDD[BostonHouse] = {
+    val myFile = spark.sparkContext.textFile(path)
 
-    myFile.filter(_.nonEmpty).zipWithIndex.map { case (x, number) =>
+    myFile.filter(_.nonEmpty).zipWithIndex.map { case (x, id) =>
       val words = x.replaceAll("\\s+", " ").replaceAll(s"^\\s+(?m)", "").replaceAll(s"(?m)\\s+$$", "").split(" ")
-      BostonHouse(number.toInt, words(0).toDouble, words(1).toDouble, words(2).toDouble, words(3), words(4).toDouble,
+      BostonHouse(id.toInt, words(0).toDouble, words(1).toDouble, words(2).toDouble, words(3), words(4).toDouble,
         words(5).toDouble, words(6).toDouble, words(7).toDouble, words(8).toInt, words(9).toDouble,
         words(10).toDouble, words(11).toDouble, words(12).toDouble, words(13).toDouble)
     }
   }
 
   val trainingReader = new CustomReader[BostonHouse](key = _.rowId.toString) {
-    def readFn(params: OpParams)(implicit spark: SparkSession): Either[RDD[BostonHouse], Dataset[BostonHouse]] = {
-      val Array(train, _) = customRead(Some(getFinalReadPath(params)), spark).randomSplit(weights = Array(0.9, 0.1),
-        seed = randomSeed)
-      Left(train)
+    def readFn(params: OpParams)(implicit spark: SparkSession): Either[RDD[BostonHouse], Dataset[BostonHouse]] = Left {
+      val Array(train, _) = customRead(getFinalReadPath(params)).randomSplit(weights = Array(0.9, 0.1), randomSeed)
+      train
     }
   }
 
   val scoringReader = new CustomReader[BostonHouse](key = _.rowId.toString) {
-    def readFn(params: OpParams)(implicit spark: SparkSession): Either[RDD[BostonHouse], Dataset[BostonHouse]] = {
-      val Array(_, test) = customRead(Some(getFinalReadPath(params)), spark).randomSplit(weights = Array(0.9, 0.1),
-        seed = randomSeed)
-      Left(test)
+    def readFn(params: OpParams)(implicit spark: SparkSession): Either[RDD[BostonHouse], Dataset[BostonHouse]] = Left {
+      val Array(_, test) = customRead(getFinalReadPath(params)).randomSplit(weights = Array(0.9, 0.1), randomSeed)
+      test
     }
   }
 
-
   ////////////////////////////////////////////////////////////////////////////////
   // WORKFLOW DEFINITION
   /////////////////////////////////////////////////////////////////////////////////
 
   val houseFeatures = Seq(crim, zn, indus, chas, nox, rm, age, dis, rad, tax, ptratio, b, lstat).transmogrify()
 
+  val splitter = DataSplitter(seed = randomSeed)
+
   val prediction = RegressionModelSelector
     .withCrossValidation(
-      dataSplitter = Some(DataSplitter(seed = randomSeed)), seed = randomSeed,
+      dataSplitter = Some(splitter), seed = randomSeed,
       modelTypesToUse = Seq(OpGBTRegressor, OpRandomForestRegressor)
     ).setInput(medv, houseFeatures).getOutput()
 

helloworld/src/main/scala/com/salesforce/hw/iris/IrisKryoRegistrator.scala → helloworld/src/main/scala/com/salesforce/hw/iris/Iris.scala

@@ -30,13 +30,11 @@
 
 package com.salesforce.hw.iris
 
-import com.esotericsoftware.kryo.Kryo
-import com.salesforce.op.utils.kryo.OpKryoRegistrator
-
-class IrisKryoRegistrator extends OpKryoRegistrator {
-
-  override def registerCustomClasses(kryo: Kryo): Unit = {
-    doAvroRegistration[com.salesforce.hw.iris.Iris](kryo)
-  }
-
-}
+case class Iris
+(
+  sepalLength: Double,
+  sepalWidth: Double,
+  petalLength: Double,
+  petalWidth: Double,
+  irisClass: String
+)

helloworld/src/main/scala/com/salesforce/hw/iris/IrisFeatures.scala

@@ -34,10 +34,9 @@ import com.salesforce.op.features.FeatureBuilder
 import com.salesforce.op.features.types._
 
 trait IrisFeatures extends Serializable {
-  val id = FeatureBuilder.Integral[Iris].extract(_.getID.toIntegral).asPredictor
-  val sepalLength = FeatureBuilder.Real[Iris].extract(_.getSepalLength.toReal).asPredictor
-  val sepalWidth = FeatureBuilder.Real[Iris].extract(_.getSepalWidth.toReal).asPredictor
-  val petalLength = FeatureBuilder.Real[Iris].extract(_.getPetalLength.toReal).asPredictor
-  val petalWidth = FeatureBuilder.Real[Iris].extract(_.getPetalWidth.toReal).asPredictor
-  val irisClass = FeatureBuilder.Text[Iris].extract(_.getClass$.toText).asResponse
+  val sepalLength = FeatureBuilder.Real[Iris].extract(_.sepalLength.toReal).asPredictor
+  val sepalWidth = FeatureBuilder.Real[Iris].extract(_.sepalWidth.toReal).asPredictor
+  val petalLength = FeatureBuilder.Real[Iris].extract(_.petalLength.toReal).asPredictor
+  val petalWidth = FeatureBuilder.Real[Iris].extract(_.petalWidth.toReal).asPredictor
+  val irisClass = FeatureBuilder.Text[Iris].extract(_.irisClass.toText).asResponse
 }

helloworld/src/main/scala/com/salesforce/hw/iris/OpIris.scala

@@ -32,37 +32,24 @@ package com.salesforce.hw.iris
 
 import com.salesforce.op._
 import com.salesforce.op.evaluators.Evaluators
-import com.salesforce.op.readers.CustomReader
+import com.salesforce.op.readers.DataReaders
 import com.salesforce.op.stages.impl.classification.MultiClassificationModelSelector
 import com.salesforce.op.stages.impl.tuning.DataCutter
-import com.salesforce.op.utils.kryo.OpKryoRegistrator
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{Dataset, SparkSession}
+import org.apache.spark.sql.Encoders
 
 /**
  * TransmogrifAI MultiClass Classification example on the Iris Dataset
  */
 object OpIris extends OpAppWithRunner with IrisFeatures {
 
-  override def kryoRegistrator: Class[_ <: OpKryoRegistrator] = classOf[IrisKryoRegistrator]
+  implicit val irisEncoder = Encoders.product[Iris]
 
   ////////////////////////////////////////////////////////////////////////////////
   // READER DEFINITIONS
   /////////////////////////////////////////////////////////////////////////////////
 
-  val randomSeed = 42
 
-  val irisReader = new CustomReader[Iris](key = _.getID.toString){
-    def readFn(params: OpParams)(implicit spark: SparkSession): Either[RDD[Iris], Dataset[Iris]] = {
-      val path = getFinalReadPath(params)
-      val myFile = spark.sparkContext.textFile(path)
-
-      Left(myFile.filter(_.nonEmpty).zipWithIndex.map { case (x, id) =>
-        val Array(sepalLength, sepalWidth, petalLength, petalWidth, klass) = x.split(",")
-        new Iris(id.toInt, sepalLength.toDouble, sepalWidth.toDouble, petalLength.toDouble, petalWidth.toDouble, klass)
-      })
-    }
-  }
+  val irisReader = DataReaders.Simple.csvCase[Iris]()
 
   ////////////////////////////////////////////////////////////////////////////////
   // WORKFLOW DEFINITION
@@ -72,6 +59,8 @@ object OpIris extends OpAppWithRunner with IrisFeatures {
 
   val features = Seq(sepalLength, sepalWidth, petalLength, petalWidth).transmogrify()
 
+  val randomSeed = 42L
+
   val cutter = DataCutter(reserveTestFraction = 0.2, seed = randomSeed)
 
   val prediction = MultiClassificationModelSelector

helloworld/src/main/scala/com/salesforce/hw/titanic/OpTitanic.scala

@@ -47,7 +47,7 @@ object OpTitanic extends OpAppWithRunner with TitanicFeatures {
   // READER DEFINITION
   /////////////////////////////////////////////////////////////////////////////////
 
-  val randomSeed = 112233L
+  val randomSeed = 42L
   val simpleReader = DataReaders.Simple.csv[Passenger](
     schema = Passenger.getClassSchema.toString, key = _.getPassengerId.toString
   )

helloworld/src/main/scala/com/salesforce/hw/titanic/TitanicFeatures.scala

@@ -34,27 +34,15 @@ import com.salesforce.op.features.FeatureBuilder
 import com.salesforce.op.features.types._
 
 trait TitanicFeatures extends Serializable {
-
   val survived = FeatureBuilder.RealNN[Passenger].extract(_.getSurvived.toDouble.toRealNN).asResponse
-
   val pClass = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getPclass).map(_.toString).toPickList).asPredictor // scalastyle:off
-
   val name = FeatureBuilder.Text[Passenger].extract(d => Option(d.getName).toText).asPredictor
-
   val sex = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getSex).toPickList).asPredictor
-
   val age = FeatureBuilder.Real[Passenger].extract(d => Option(Double.unbox(d.getAge)).toReal).asPredictor
-
   val sibSp = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getSibSp).map(_.toString).toPickList).asPredictor
-
   val parch = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getParch).map(_.toString).toPickList).asPredictor
-
   val ticket = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getTicket).toPickList).asPredictor
-
   val fare = FeatureBuilder.Real[Passenger].extract(d => Option(Double.unbox(d.getFare)).toReal).asPredictor
-
   val cabin = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getCabin).toPickList).asPredictor
-
   val embarked = FeatureBuilder.PickList[Passenger].extract(d => Option(d.getEmbarked).toPickList).asPredictor
-
 }