Model combiner

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

@@ -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/stages/impl/preparators/SanityCheckerMetadata.scala

@@ -179,6 +179,9 @@ case class SummaryStatistics
  meta.build()
  }
 
+ private[op] def +(sum: SummaryStatistics): SummaryStatistics = new SummaryStatistics(count, sampleFraction,
+ max ++ sum.max, min ++ sum.min, mean ++ sum.mean, variance ++ sum.variance)
+
 }
 
 /**
@@ -313,10 +316,22 @@ case class Correlations
  corrMeta.putString(SanityCheckerNames.CorrelationType, corrType.sparkName)
  corrMeta.build()
  }
+
+ private[op] def +(corr: Correlations): Correlations =
+ new Correlations(featuresIn ++ corr.featuresIn, values ++ corr.values, nanCorrs ++ corr.nanCorrs, corrType)
 }
 
 case object SanityCheckerSummary {
 
+ def flatten(checkers: Seq[SanityCheckerSummary]): SanityCheckerSummary = {
+ val correlationsWLabel: Correlations = checkers.map(_.correlationsWLabel).reduce(_ + _)
+ val dropped: Seq[String] = checkers.flatMap(_.dropped)
+ val featuresStatistics: SummaryStatistics = checkers.map(_.featuresStatistics).reduce(_ + _)
+ val names: Seq[String] = checkers.flatMap(_.names)
+ val categoricalStats: Array[CategoricalGroupStats] = checkers.flatMap(_.categoricalStats).toArray
+ new SanityCheckerSummary(correlationsWLabel, dropped, featuresStatistics, names, categoricalStats)
+ }
+
  private def correlationsFromMetadata(meta: Metadata): Correlations = {
  val wrapped = meta.wrapped
  Correlations(