Model combiner

core/src/main/scala/com/salesforce/op/ModelInsights.scala

@@ -35,7 +35,7 @@ import com.salesforce.op.features._
 import com.salesforce.op.features.types._
 import com.salesforce.op.filters._
 import com.salesforce.op.stages._
-import com.salesforce.op.stages.impl.feature.{TextStats, TransmogrifierDefaults}
+import com.salesforce.op.stages.impl.feature.{CombinationStrategy, TextStats, TransmogrifierDefaults}
 import com.salesforce.op.stages.impl.preparators._
 import com.salesforce.op.stages.impl.selector._
 import com.salesforce.op.stages.impl.tuning.{DataBalancerSummary, DataCutterSummary, DataSplitterSummary}
@@ -46,6 +46,7 @@ import com.salesforce.op.utils.spark.RichMetadata._
 import com.salesforce.op.utils.spark.{OpVectorColumnMetadata, OpVectorMetadata}
 import com.salesforce.op.utils.table.Alignment._
 import com.salesforce.op.utils.table.Table
+import com.twitter.algebird.Operators._
 import com.twitter.algebird.Moments
 import ml.dmlc.xgboost4j.scala.spark.OpXGBoost.RichBooster
 import ml.dmlc.xgboost4j.scala.spark.{XGBoostClassificationModel, XGBoostRegressionModel}
@@ -443,43 +444,69 @@ case object ModelInsights {
  blacklistedMapKeys: Map[String, Set[String]],
  rawFeatureFilterResults: RawFeatureFilterResults
  ): ModelInsights = {
- val sanityCheckers = stages.collect { case s: SanityCheckerModel => s }
- val sanityChecker = sanityCheckers.lastOption
- val checkerSummary = sanityChecker.map(s => SanityCheckerSummary.fromMetadata(s.getMetadata().getSummaryMetadata()))
- log.info(
- s"Found ${sanityCheckers.length} sanity checkers will " +
- s"${sanityChecker.map("use results from the last checker:" + _.uid + "to").getOrElse("not")}" +
- s" to fill in model insights"
- )
 
+ // TODO support other model types?
  val models = stages.collect{
  case s: SelectedModel => s
  case s: OpPredictorWrapperModel[_] => s
- } // TODO support other model types?
- val model = models.lastOption
+ case s: SelectedCombinerModel => s
+ }
+ val model = models.flatMap{
+ case s: SelectedCombinerModel if s.strategy == CombinationStrategy.Best =>
+ val originF = if (s.weight1 > 0.5) s.getInputFeature[Prediction](1) else s.getInputFeature[Prediction](2)
+ models.find( m => originF.exists(_.originStage.uid == m.uid) )
+ case s => Option(s)
+ }.lastOption
+
  log.info(
  s"Found ${models.length} models will " +
  s"${model.map("use results from the last model:" + _.uid + "to").getOrElse("not")}" +
  s" to fill in model insights"
  )
 
- val label = model.map(_.getInputFeature[RealNN](0)).orElse(sanityChecker.map(_.getInputFeature[RealNN](0))).flatten
+ val modelInputStages: Set[String] = model.map { m =>
+ val stages = m.getInputFeatures().map(_.parentStages().toOption.map(_.keySet.map(_.uid)))
+ val uid = stages.collect{ case Some(uids) => uids }
+ uid.fold(Set.empty)(_ + _)
+ }.getOrElse(Set.empty)
+
+ val sanityCheckers = stages.collect { case s: SanityCheckerModel => s }
+ val sanityCheckersForModel = sanityCheckers.filter(s => modelInputStages.contains(s.uid) &&
+ model.exists(_.getInputFeature[RealNN](0) == s.getInputFeature[RealNN](0))).toSeq
+
+ val sanityChecker = if (sanityCheckersForModel.nonEmpty) sanityCheckersForModel else sanityCheckers.lastOption.toSeq
+ val checkerSummary = if (sanityChecker.nonEmpty) {
+ Option(SanityCheckerSummary.flatten(
+ sanityChecker.map(s => SanityCheckerSummary.fromMetadata(s.getMetadata().getSummaryMetadata()))
+ ))
+ } else None
+
+ log.info(
+ s"Found ${sanityCheckers.length} sanity checkers" +
+ s"${sanityChecker.map("will preferentially use results from checkers in model path:" + _.uid +
+ " to fill in model insights")}"
+ )
+
+
+ val label = model.map(_.getInputFeature[RealNN](0))
+ .orElse(sanityChecker.lastOption.map(_.getInputFeature[RealNN](0))).flatten
  log.info(s"Found ${label.map(_.name + " as label").getOrElse("no label")} to fill in model insights")
 
  // Recover the vector metadata
  val vectorInput: Option[OpVectorMetadata] = {
  def makeMeta(s: => OpPipelineStageParams) = Try(OpVectorMetadata(s.getInputSchema().last)).toOption
 
- sanityChecker
- // first try out to get vector metadata from sanity checker
- .flatMap(s => makeMeta(s.parent.asInstanceOf[SanityChecker]).orElse(makeMeta(s)))
- // fall back to model selector stage metadata
- .orElse(model.flatMap(m => makeMeta(m.parent.asInstanceOf[ModelSelector[_, _]])))
- // finally try to get it from the last vector stage
- .orElse(
- stages.filter(_.getOutput().isSubtypeOf[OPVector]).lastOption
- .map(v => OpVectorMetadata(v.getOutputFeatureName, v.getMetadata()))
- )
+ if (sanityChecker.nonEmpty) { // first try out to get vector metadata from sanity checker
+ Option(OpVectorMetadata.flatten("",
+ sanityChecker.flatMap(s => makeMeta(s.parent.asInstanceOf[SanityChecker]).orElse(makeMeta(s))))
+ )
+ } else {
+ model.flatMap(m => makeMeta(m)) // fall back to model selector stage metadata
+ .orElse( // finally try to get it from the last vector stage
+ stages.filter(_.getOutput().isSubtypeOf[OPVector]).lastOption
+ .map(v => OpVectorMetadata(v.getOutputFeatureName, v.getMetadata()))
+ )
+ }
  }
  log.info(
  s"Found ${vectorInput.map(_.name + " as feature vector").getOrElse("no feature vector")}" +
@@ -639,7 +666,7 @@ case object ModelInsights {
  contribution =
  contributions.map(_.applyOrElse(h.index, (_: Int) => 0.0)) // nothing dropped without sanity check
  )
- }
+  }
  case (None, _) => Seq.empty
  }
 
@@ -712,9 +739,9 @@ case object ModelInsights {
  // need to also divide by labelStd for linear regression
  // See https://u.demog.berkeley.edu/~andrew/teaching/standard_coeff.pdf
  // See https://en.wikipedia.org/wiki/Standardized_coefficient
- sparkFtrContrib.map(_ * featureStd / labelStd)
- }
- else sparkFtrContrib
+  sparkFtrContrib.map(_ * featureStd / labelStd)
+  }
+  else sparkFtrContrib
  case _ => sparkFtrContrib
  }
  }
@@ -747,6 +774,8 @@ case object ModelInsights {
  model match {
  case Some(m: SelectedModel) =>
  Try(ModelSelectorSummary.fromMetadata(m.getMetadata().getSummaryMetadata())).toOption
+ case Some(m: SelectedCombinerModel) =>
+ Try(ModelSelectorSummary.fromMetadata(m.getMetadata().getSummaryMetadata())).toOption
  case _ => None
  }
  }

core/src/main/scala/com/salesforce/op/evaluators/EvaluationMetrics.scala

@@ -99,20 +99,25 @@ sealed trait EvalMetric extends EnumEntry with Serializable {
  */
  def humanFriendlyName: String
 
+ /**
+ * Is this metric being larger better or is smaller better
+ */
+ def isLargerBetter: Boolean
+
 }
 
 /**
  * Eval metric companion object
  */
 object EvalMetric {
 
- def withNameInsensitive(name: String): EvalMetric = {
+ def withNameInsensitive(name: String, isLargerBetter: Boolean = true): EvalMetric = {
  BinaryClassEvalMetrics.withNameInsensitiveOption(name)
  .orElse(MultiClassEvalMetrics.withNameInsensitiveOption(name))
  .orElse(RegressionEvalMetrics.withNameInsensitiveOption(name))
  .orElse(ForecastEvalMetrics.withNameInsensitiveOption(name))
  .orElse(OpEvaluatorNames.withNameInsensitiveOption(name))
- .getOrElse(OpEvaluatorNames.Custom(name, name))
+ .getOrElse(OpEvaluatorNames.Custom(name, name, isLargerBetter))
  }
 }
 
@@ -122,37 +127,38 @@ object EvalMetric {
 sealed abstract class ClassificationEvalMetric
 (
  val sparkEntryName: String,
- val humanFriendlyName: String
+ val humanFriendlyName: String,
+ val isLargerBetter: Boolean
 ) extends EvalMetric
 
 /**
  * Binary Classification Metrics
  */
 object BinaryClassEvalMetrics extends Enum[ClassificationEvalMetric] {
  val values = findValues
- case object Precision extends ClassificationEvalMetric("weightedPrecision", "precision")
- case object Recall extends ClassificationEvalMetric("weightedRecall", "recall")
- case object F1 extends ClassificationEvalMetric("f1", "f1")
- case object Error extends ClassificationEvalMetric("accuracy", "error")
- case object AuROC extends ClassificationEvalMetric("areaUnderROC", "area under ROC")
- case object AuPR extends ClassificationEvalMetric("areaUnderPR", "area under precision-recall")
- case object TP extends ClassificationEvalMetric("TP", "true positive")
- case object TN extends ClassificationEvalMetric("TN", "true negative")
- case object FP extends ClassificationEvalMetric("FP", "false positive")
- case object FN extends ClassificationEvalMetric("FN", "false negative")
- case object BrierScore extends ClassificationEvalMetric("brierScore", "brier score")
+ case object Precision extends ClassificationEvalMetric("weightedPrecision", "precision", true)
+ case object Recall extends ClassificationEvalMetric("weightedRecall", "recall", true)
+ case object F1 extends ClassificationEvalMetric("f1", "f1", true)
+ case object Error extends ClassificationEvalMetric("accuracy", "error", false)
+ case object AuROC extends ClassificationEvalMetric("areaUnderROC", "area under ROC", true)
+ case object AuPR extends ClassificationEvalMetric("areaUnderPR", "area under precision-recall", true)
+ case object TP extends ClassificationEvalMetric("TP", "true positive", true)
+ case object TN extends ClassificationEvalMetric("TN", "true negative", true)
+ case object FP extends ClassificationEvalMetric("FP", "false positive", false)
+ case object FN extends ClassificationEvalMetric("FN", "false negative", false)
+ case object BrierScore extends ClassificationEvalMetric("brierScore", "brier score", false)
 }
 
 /**
  * Multi Classification Metrics
  */
 object MultiClassEvalMetrics extends Enum[ClassificationEvalMetric] {
  val values = findValues
- case object Precision extends ClassificationEvalMetric("weightedPrecision", "precision")
- case object Recall extends ClassificationEvalMetric("weightedRecall", "recall")
- case object F1 extends ClassificationEvalMetric("f1", "f1")
- case object Error extends ClassificationEvalMetric("accuracy", "error")
- case object ThresholdMetrics extends ClassificationEvalMetric("thresholdMetrics", "threshold metrics")
+ case object Precision extends ClassificationEvalMetric("weightedPrecision", "precision", true)
+ case object Recall extends ClassificationEvalMetric("weightedRecall", "recall", true)
+ case object F1 extends ClassificationEvalMetric("f1", "f1", true)
+ case object Error extends ClassificationEvalMetric("accuracy", "error", false)
+ case object ThresholdMetrics extends ClassificationEvalMetric("thresholdMetrics", "threshold metrics", true)
 }
 
 
@@ -162,18 +168,19 @@ object MultiClassEvalMetrics extends Enum[ClassificationEvalMetric] {
 sealed abstract class RegressionEvalMetric
 (
  val sparkEntryName: String,
- val humanFriendlyName: String
+ val humanFriendlyName: String,
+ val isLargerBetter: Boolean
 ) extends EvalMetric
 
 /**
  * Regression Metrics
  */
 object RegressionEvalMetrics extends Enum[RegressionEvalMetric] {
  val values: Seq[RegressionEvalMetric] = findValues
- case object RootMeanSquaredError extends RegressionEvalMetric("rmse", "root mean square error")
- case object MeanSquaredError extends RegressionEvalMetric("mse", "mean square error")
- case object R2 extends RegressionEvalMetric("r2", "r2")
- case object MeanAbsoluteError extends RegressionEvalMetric("mae", "mean absolute error")
+ case object RootMeanSquaredError extends RegressionEvalMetric("rmse", "root mean square error", false)
+ case object MeanSquaredError extends RegressionEvalMetric("mse", "mean square error", false)
+ case object R2 extends RegressionEvalMetric("r2", "r2", true)
+ case object MeanAbsoluteError extends RegressionEvalMetric("mae", "mean absolute error", false)
 }
 
 
@@ -183,15 +190,16 @@ object RegressionEvalMetrics extends Enum[RegressionEvalMetric] {
 sealed abstract class ForecastEvalMetric
 (
  val sparkEntryName: String,
- val humanFriendlyName: String
+ val humanFriendlyName: String,
+ val isLargerBetter: Boolean
 ) extends EvalMetric
 
 
 object ForecastEvalMetrics extends Enum[ForecastEvalMetric] {
  val values: Seq[ForecastEvalMetric] = findValues
- case object SMAPE extends ForecastEvalMetric("smape", "symmetric mean absolute percentage error")
- case object MASE extends ForecastEvalMetric("mase", "mean absolute scaled error")
- case object SeasonalError extends ForecastEvalMetric("seasonalError", "seasonal error")
+ case object SMAPE extends ForecastEvalMetric("smape", "symmetric mean absolute percentage error", false)
+ case object MASE extends ForecastEvalMetric("mase", "mean absolute scaled error", false)
+ case object SeasonalError extends ForecastEvalMetric("seasonalError", "seasonal error", false)
 }
 
 
@@ -201,25 +209,32 @@ object ForecastEvalMetrics extends Enum[ForecastEvalMetric] {
 sealed abstract class OpEvaluatorNames
 (
  val sparkEntryName: String,
- val humanFriendlyName: String
+ val humanFriendlyName: String,
+ val isLargerBetter: Boolean // for default value
 ) extends EvalMetric
 
 /**
  * Contains evaluator names used in logging
  */
 object OpEvaluatorNames extends Enum[OpEvaluatorNames] {
  val values: Seq[OpEvaluatorNames] = findValues
- case object Binary extends OpEvaluatorNames("binEval", "binary evaluation metrics")
- case object BinScore extends OpEvaluatorNames("binScoreEval", "bin score evaluation metrics")
- case object Multi extends OpEvaluatorNames("multiEval", "multiclass evaluation metrics")
- case object Regression extends OpEvaluatorNames("regEval", "regression evaluation metrics")
- case object Forecast extends OpEvaluatorNames("regForecast", "forecast evaluation metrics")
- case class Custom(name: String, humanName: String) extends OpEvaluatorNames(name, humanName) {
+ case object Binary extends OpEvaluatorNames("binEval", "binary evaluation metrics", true)
+ case object BinScore extends OpEvaluatorNames("binScoreEval", "bin score evaluation metrics", false)
+ case object Multi extends OpEvaluatorNames("multiEval", "multiclass evaluation metrics", true)
+ case object Regression extends OpEvaluatorNames("regEval", "regression evaluation metrics", false)
+ case object Forecast extends OpEvaluatorNames("regForecast", "regression evaluation metrics", false)
+ case class Custom(name: String, humanName: String, largeBetter: Boolean) extends
+ OpEvaluatorNames(name, humanName, largeBetter) {
  override def entryName: String = name.toLowerCase
  }
+ def withName(name: String, isLargerBetter: Boolean): OpEvaluatorNames =
+ Try(super.withName(name)).getOrElse(Custom(name, name, isLargerBetter))
+ override def withName(name: String): OpEvaluatorNames = withName(name, true)
+
+ def withNameInsensitive(name: String, isLargerBetter: Boolean): OpEvaluatorNames =
+ super.withNameInsensitiveOption(name).getOrElse(Custom(name, name, isLargerBetter))
+ override def withNameInsensitive(name: String): OpEvaluatorNames = withNameInsensitive(name, true)
+
  def withFriendlyNameInsensitive(name: String): Option[OpEvaluatorNames] =
  values.collectFirst { case n if n.humanFriendlyName.equalsIgnoreCase(name) => n }
- override def withName(name: String): OpEvaluatorNames = Try(super.withName(name)).getOrElse(Custom(name, name))
- override def withNameInsensitive(name: String): OpEvaluatorNames = super.withNameInsensitiveOption(name)
- .getOrElse(Custom(name, name))
 }