Spark 2.4 support

README.md

@@ -128,7 +128,7 @@ Start by picking TransmogrifAI version to match your project dependencies from t
 
 | TransmogrifAI Version                                 | Spark Version | Scala Version | Java Version |
 |-------------------------------------------------------|:-------------:|:-------------:|:------------:|
-| 0.6.2 (unreleased, master)                            |      2.3      |      2.11     |      1.8     |
+| 0.7.0 (unreleased, master)                            |      2.4      |      2.11     |      1.8     |
 | **0.6.1 (stable)**, 0.6.0, 0.5.3, 0.5.2, 0.5.1, 0.5.0 |    **2.3**    |    **2.11**   |    **1.8**   |
 | 0.4.0, 0.3.4                                          |      2.2      |      2.11     |      1.8     |
 

build.gradle

@@ -1,15 +1,16 @@
 buildscript {
     repositories {
-        maven { url "https://plugins.gradle.org/m2/" }
         mavenCentral()
+        jcenter()
+        maven { url "https://plugins.gradle.org/m2/" }
     }
     dependencies {
         classpath 'org.github.ngbinh.scalastyle:gradle-scalastyle-plugin_2.11:1.0.1'
+        classpath 'com.commercehub.gradle.plugin:gradle-avro-plugin:0.16.0'
     }
 }
 
 plugins {
-    id 'com.commercehub.gradle.plugin.avro' version '0.8.0'
     id 'org.scoverage' version '2.5.0'
     id 'net.minecrell.licenser' version '0.4.1'
     id 'com.github.jk1.dependency-license-report' version '0.5.0'
@@ -57,14 +58,13 @@ configure(allProjs) {
         scalaVersionRevision = '12'
         scalaTestVersion = '3.0.5'
         scalaCheckVersion = '1.14.0'
-        junitVersion = '4.11'
-        avroVersion = '1.7.7'
-        sparkVersion = '2.3.2'
-        sparkAvroVersion = '4.0.0'
+        junitVersion = '4.12'
+        avroVersion = '1.8.2'
+        sparkVersion = '2.4.4'
         scalaGraphVersion = '1.12.5'
         scalafmtVersion = '1.5.1'
         hadoopVersion = 'hadoop2'
-        json4sVersion = '3.2.11' // matches Spark dependency version
+        json4sVersion = '3.5.3' // matches Spark dependency version
         jodaTimeVersion = '2.9.4'
         jodaConvertVersion = '1.8.1'
         algebirdVersion = '0.13.4'
@@ -75,20 +75,20 @@ configure(allProjs) {
         googleLibPhoneNumberVersion = '8.8.5'
         googleGeoCoderVersion = '2.82'
         googleCarrierVersion = '1.72'
-        chillVersion = '0.8.4'
+        chillVersion = '0.9.3'
         reflectionsVersion = '0.9.11'
         collectionsVersion = '3.2.2'
         optimaizeLangDetectorVersion = '0.0.1'
         tikaVersion = '1.22'
-        sparkTestingBaseVersion = '2.3.1_0.10.0'
+        sparkTestingBaseVersion = '2.4.3_0.12.0'
         sourceCodeVersion = '0.1.3'
         pegdownVersion = '1.4.2'
         commonsValidatorVersion = '1.6'
         commonsIOVersion = '2.6'
         scoveragePluginVersion = '1.3.1'
-        xgboostVersion = '0.81'
+        xgboostVersion = '0.90'
         akkaSlf4jVersion = '2.3.11'
-        mleapVersion = '0.13.0'
+        mleapVersion = '0.14.0'
         memoryFilesystemVersion = '2.1.0'
     }
 

cli/build.gradle

@@ -69,20 +69,16 @@ task copyTemplates(type: Copy) {
         fileName.replace(".gradle.template", ".gradle")
     }
     expand([
-            databaseHostname: 'db.company.com',
             version: scalaVersion,
             scalaVersion: scalaVersion,
             scalaVersionRevision: scalaVersionRevision,
             scalaTestVersion: scalaTestVersion,
             junitVersion: junitVersion,
             sparkVersion: sparkVersion,
             avroVersion: avroVersion,
-            sparkAvroVersion: sparkAvroVersion,
             hadoopVersion: hadoopVersion,
             collectionsVersion: collectionsVersion,
-            transmogrifaiVersion: version,
-            buildNumber: (int)(Math.random() * 1000),
-            date: new Date()
+            transmogrifaiVersion: version
     ])
 }
 

cli/src/main/scala/com/salesforce/op/cli/SchemaSource.scala

@@ -138,7 +138,7 @@ case class AutomaticSchema(recordClassName: String)(dataFile: File) extends Sche
         case Some(actualType) =>
           val newSchema = Schema.create(actualType)
           val schemaField =
-            new Schema.Field(field.name, newSchema, "auto-generated", orgSchemaField.defaultValue)
+            new Schema.Field(field.name, newSchema, "auto-generated", orgSchemaField.defaultVal())
           AvroField.from(schemaField)
       }
     } else field

cli/src/test/scala/com/salesforce/op/cli/gen/AvroFieldTest.scala

@@ -69,7 +69,7 @@ class AvroFieldTest extends FlatSpec with TestCommon with Assertions {
     val allSchemas = (enum::unions)++simpleSchemas // NULL does not work
 
     val fields = allSchemas.zipWithIndex map {
-      case (s, i) => new Schema.Field("x" + i, s, "Who", null)
+      case (s, i) => new Schema.Field("x" + i, s, "Who", null: Object)
     }
 
     val expected = List(
@@ -86,7 +86,7 @@ class AvroFieldTest extends FlatSpec with TestCommon with Assertions {
 
     an[IllegalArgumentException] should be thrownBy {
       val nullSchema = Schema.create(Schema.Type.NULL)
-      val nullField = new Schema.Field("xxx", null, "Nobody", null)
+      val nullField = new Schema.Field("xxx", null, "Nobody", null: Object)
       AvroField from nullField
     }
 

core/src/main/scala/com/salesforce/op/stages/impl/classification/OpLinearSVC.scala

@@ -152,14 +152,14 @@ class OpLinearSVCModel
 ) extends OpPredictorWrapperModel[LinearSVCModel](uid = uid, operationName = operationName, sparkModel = sparkModel) {
 
   @transient lazy private val predictRaw = reflectMethod(getSparkMlStage().get, "predictRaw")
-  @transient lazy private val predict = reflectMethod(getSparkMlStage().get, "predict")
+  @transient lazy private val predict: Vector => Double = getSparkMlStage().get.predict(_)
 
   /**
    * Function used to convert input to output
    */
   override def transformFn: (RealNN, OPVector) => Prediction = (label, features) => {
     val raw = predictRaw(features.value).asInstanceOf[Vector]
-    val pred = predict(features.value).asInstanceOf[Double]
+    val pred = predict(features.value)
 
     Prediction(rawPrediction = raw, prediction = pred)
   }

core/src/main/scala/com/salesforce/op/stages/impl/classification/OpXGBoostClassifier.scala

@@ -235,6 +235,11 @@ class OpXGBoostClassifier(uid: String = UID[OpXGBoostClassifier])
    */
   def setMaxBins(value: Int): this.type = set(maxBins, value)
 
+  /**
+   * Maximum number of nodes to be added. Only relevant when grow_policy=lossguide is set.
+   */
+  def setMaxLeaves(value: Int): this.type = set(maxLeaves, value)
+
   /**
    * This is only used for approximate greedy algorithm.
    * This roughly translated into O(1 / sketch_eps) number of bins. Compared to directly select
@@ -282,8 +287,19 @@ class OpXGBoostClassifier(uid: String = UID[OpXGBoostClassifier])
   def setLambdaBias(value: Double): this.type = set(lambdaBias, value)
 
   // setters for learning params
+
+  /**
+   * Specify the learning task and the corresponding learning objective.
+   * options: reg:squarederror, reg:logistic, binary:logistic, binary:logitraw, count:poisson,
+   * multi:softmax, multi:softprob, rank:pairwise, reg:gamma. default: reg:squarederror
+   */
   def setObjective(value: String): this.type = set(objective, value)
 
+  /**
+   * Objective type used for training. For options see [[ml.dmlc.xgboost4j.scala.spark.params.LearningTaskParams]]
+   */
+  def setObjectiveType(value: String): this.type = set(objectiveType, value)
+
   /**
    * Specify the learning task and the corresponding learning objective.
    * options: reg:linear, reg:logistic, binary:logistic, binary:logitraw, count:poisson,
@@ -310,6 +326,11 @@ class OpXGBoostClassifier(uid: String = UID[OpXGBoostClassifier])
    */
   def setNumEarlyStoppingRounds(value: Int): this.type = set(numEarlyStoppingRounds, value)
 
+  /**
+   * Define the expected optimization to the evaluation metrics, true to maximize otherwise minimize it
+   */
+  def setMaximizeEvaluationMetrics(value: Boolean): this.type = set(maximizeEvaluationMetrics, value)
+
   /**
    * Customized objective function provided by user. default: null
    */
@@ -359,17 +380,18 @@ class OpXGBoostClassificationModel
 
   private lazy val model = getSparkMlStage().get
   private lazy val booster = model.nativeBooster
-  private lazy val treeLimit = model.getTreeLimit.toInt
+  private lazy val treeLimit = model.getTreeLimit
   private lazy val missing = model.getMissing
 
   override def transformFn: (RealNN, OPVector) => Prediction = (label, features) => {
-    val data = removeMissingValues(Iterator(features.value.asXGB), missing)
+    val data = processMissingValues(Iterator(features.value.asXGB), missing)
     val dm = new DMatrix(dataIter = data)
     val rawPred = booster.predict(dm, outPutMargin = true, treeLimit = treeLimit)(0).map(_.toDouble)
     val rawPrediction = if (model.numClasses == 2) Array(-rawPred(0), rawPred(0)) else rawPred
     val prob = booster.predict(dm, outPutMargin = false, treeLimit = treeLimit)(0).map(_.toDouble)
     val probability = if (model.numClasses == 2) Array(1.0 - prob(0), prob(0)) else prob
     val prediction = probability2predictionMirror(Vectors.dense(probability)).asInstanceOf[Double]
+
     Prediction(prediction = prediction, rawPrediction = rawPrediction, probability = probability)
   }
 }

core/src/main/scala/com/salesforce/op/stages/impl/regression/OpDecisionTreeRegressor.scala

@@ -34,7 +34,6 @@ import com.salesforce.op.UID
 import com.salesforce.op.features.types.{OPVector, Prediction, RealNN}
 import com.salesforce.op.stages.impl.CheckIsResponseValues
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictionModel, OpPredictorWrapper}
-import com.salesforce.op.utils.reflection.ReflectionUtils.reflectMethod
 import org.apache.spark.ml.regression.{DecisionTreeRegressionModel, DecisionTreeRegressor, OpDecisionTreeRegressorParams}
 
 import scala.reflect.runtime.universe.TypeTag
@@ -113,7 +112,4 @@ class OpDecisionTreeRegressionModel
   ttov: TypeTag[Prediction#Value]
 ) extends OpPredictionModel[DecisionTreeRegressionModel](
   sparkModel = sparkModel, uid = uid, operationName = operationName
-) {
-  @transient lazy val predictMirror = reflectMethod(getSparkMlStage().get, "predict")
-}
-
+)

core/src/main/scala/com/salesforce/op/stages/impl/regression/OpGBTRegressor.scala

@@ -34,7 +34,6 @@ import com.salesforce.op.UID
 import com.salesforce.op.features.types.{OPVector, Prediction, RealNN}
 import com.salesforce.op.stages.impl.CheckIsResponseValues
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictionModel, OpPredictorWrapper}
-import com.salesforce.op.utils.reflection.ReflectionUtils.reflectMethod
 import org.apache.spark.ml.regression.{GBTRegressionModel, GBTRegressor, OpGBTRegressorParams}
 
 import scala.reflect.runtime.universe.TypeTag
@@ -139,7 +138,4 @@ class OpGBTRegressionModel
   ttov: TypeTag[Prediction#Value]
 ) extends OpPredictionModel[GBTRegressionModel](
   sparkModel = sparkModel, uid = uid, operationName = operationName
-) {
-  @transient lazy val predictMirror = reflectMethod(getSparkMlStage().get, "predict")
-}
-
+)

core/src/main/scala/com/salesforce/op/stages/impl/regression/OpLinearRegression.scala

@@ -34,7 +34,6 @@ import com.salesforce.op._
 import com.salesforce.op.features.types.{OPVector, Prediction, RealNN}
 import com.salesforce.op.stages.impl.CheckIsResponseValues
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictionModel, OpPredictorWrapper}
-import com.salesforce.op.utils.reflection.ReflectionUtils.reflectMethod
 import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel, OpLinearRegressionParams}
 
 import scala.reflect.runtime.universe.TypeTag
@@ -205,7 +204,4 @@ class OpLinearRegressionModel
   ttov: TypeTag[Prediction#Value]
 ) extends OpPredictionModel[LinearRegressionModel](
   sparkModel = sparkModel, uid = uid, operationName = operationName
-) {
-  @transient lazy val predictMirror = reflectMethod(getSparkMlStage().get, "predict")
-}
-
+)

core/src/main/scala/com/salesforce/op/stages/impl/regression/OpRandomForestRegressor.scala

@@ -34,7 +34,6 @@ import com.salesforce.op.UID
 import com.salesforce.op.features.types.{OPVector, Prediction, RealNN}
 import com.salesforce.op.stages.impl.CheckIsResponseValues
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictionModel, OpPredictorWrapper}
-import com.salesforce.op.utils.reflection.ReflectionUtils.reflectMethod
 import org.apache.spark.ml.regression.{OpRandomForestRegressorParams, RandomForestRegressionModel, RandomForestRegressor}
 
 import scala.reflect.runtime.universe.TypeTag
@@ -126,8 +125,4 @@ class OpRandomForestRegressionModel
   ttov: TypeTag[Prediction#Value]
 ) extends OpPredictionModel[RandomForestRegressionModel](
   sparkModel = sparkModel, uid = uid, operationName = operationName
-) {
-  @transient lazy val predictMirror = reflectMethod(getSparkMlStage().get, "predict")
-}
-
-
+)

core/src/main/scala/com/salesforce/op/stages/impl/regression/OpXGBoostRegressor.scala

@@ -34,7 +34,6 @@ import com.salesforce.op.UID
 import com.salesforce.op.features.types.{OPVector, Prediction, RealNN}
 import com.salesforce.op.stages.impl.CheckIsResponseValues
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictionModel, OpPredictorWrapper}
-import com.salesforce.op.utils.reflection.ReflectionUtils.reflectMethod
 import ml.dmlc.xgboost4j.scala.{EvalTrait, ObjectiveTrait}
 import ml.dmlc.xgboost4j.scala.spark.{OpXGBoostRegressorParams, TrackerConf, XGBoostRegressionModel, XGBoostRegressor}
 
@@ -234,6 +233,11 @@ class OpXGBoostRegressor(uid: String = UID[OpXGBoostRegressor])
    */
   def setMaxBins(value: Int): this.type = set(maxBins, value)
 
+  /**
+   * Maximum number of nodes to be added. Only relevant when grow_policy=lossguide is set.
+   */
+  def setMaxLeaves(value: Int): this.type = set(maxLeaves, value)
+
   /**
    * This is only used for approximate greedy algorithm.
    * This roughly translated into O(1 / sketch_eps) number of bins. Compared to directly select
@@ -281,8 +285,19 @@ class OpXGBoostRegressor(uid: String = UID[OpXGBoostRegressor])
   def setLambdaBias(value: Double): this.type = set(lambdaBias, value)
 
   // setters for learning params
+
+  /**
+   * Specify the learning task and the corresponding learning objective.
+   * options: reg:squarederror, reg:logistic, binary:logistic, binary:logitraw, count:poisson,
+   * multi:softmax, multi:softprob, rank:pairwise, reg:gamma. default: reg:squarederror
+   */
   def setObjective(value: String): this.type = set(objective, value)
 
+  /**
+   * Objective type used for training. For options see [[ml.dmlc.xgboost4j.scala.spark.params.LearningTaskParams]]
+   */
+  def setObjectiveType(value: String): this.type = set(objectiveType, value)
+
   /**
    * Specify the learning task and the corresponding learning objective.
    * options: reg:linear, reg:logistic, binary:logistic, binary:logitraw, count:poisson,
@@ -309,6 +324,11 @@ class OpXGBoostRegressor(uid: String = UID[OpXGBoostRegressor])
    */
   def setNumEarlyStoppingRounds(value: Int): this.type = set(numEarlyStoppingRounds, value)
 
+  /**
+   * Define the expected optimization to the evaluation metrics, true to maximize otherwise minimize it
+   */
+  def setMaximizeEvaluationMetrics(value: Boolean): this.type = set(maximizeEvaluationMetrics, value)
+
   /**
    * Customized objective function provided by user. default: null
    */
@@ -341,6 +361,4 @@ class OpXGBoostRegressionModel
   ttov: TypeTag[Prediction#Value]
 ) extends OpPredictionModel[XGBoostRegressionModel](
   sparkModel = sparkModel, uid = uid, operationName = operationName
-) {
-  @transient lazy val predictMirror = reflectMethod(getSparkMlStage().get, "predict")
-}
+)

core/src/main/scala/com/salesforce/op/stages/impl/selector/ModelSelector.scala

@@ -41,7 +41,6 @@ import com.salesforce.op.stages.impl.tuning.{BestEstimator, _}
 import com.salesforce.op.stages.sparkwrappers.generic.SparkWrapperParams
 import com.salesforce.op.stages.sparkwrappers.specific.{OpPredictorWrapperModel, SparkModelConverter}
 import com.salesforce.op.utils.spark.RichMetadata._
-import com.salesforce.op.utils.spark.RichDataset._
 import com.salesforce.op.utils.spark.RichParamMap._
 import com.salesforce.op.utils.stages.FitStagesUtil._
 import org.apache.spark.ml.param._

core/src/main/scala/com/salesforce/op/stages/sparkwrappers/specific/OpPredictionModel.scala

@@ -52,9 +52,12 @@ abstract class OpPredictionModel[T <: PredictionModel[Vector, T]]
   operationName: String
 ) extends OpPredictorWrapperModel[T](uid = uid, operationName = operationName, sparkModel = sparkModel) {
 
-  protected def predictMirror: MethodMirror
-
-  protected def predict(features: Vector): Double = predictMirror.apply(features).asInstanceOf[Double]
+  /**
+   * Predict label for the given features
+   */
+  @transient protected lazy val predict: Vector => Double = getSparkMlStage().getOrElse(
+    throw new RuntimeException(s"Could not find the wrapped Spark stage.")
+  ).predict(_)
 
   /**
    * Function used to convert input to output

core/src/main/scala/ml/dmlc/xgboost4j/scala/spark/XGBoostParams.scala

@@ -104,8 +104,8 @@ case object OpXGBoost {
   }
 
   /**
-   * Hack to access [[ml.dmlc.xgboost4j.scala.spark.XGBoost.removeMissingValues]] private method
+   * Hack to access [[ml.dmlc.xgboost4j.scala.spark.XGBoost.processMissingValues]] private method
    */
-  def removeMissingValues(xgbLabelPoints: Iterator[LabeledPoint], missing: Float): Iterator[LabeledPoint] =
-    XGBoost.removeMissingValues(xgbLabelPoints, missing)
+  def processMissingValues(xgbLabelPoints: Iterator[LabeledPoint], missing: Float): Iterator[LabeledPoint] =
+    XGBoost.processMissingValues(xgbLabelPoints, missing)
 }