Detect and remove IDs disguised in text features

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.{CombinationStrategy, TextStats, TransmogrifierDefaults}
+import com.salesforce.op.stages.impl.feature.{CombinationStrategy, 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}

core/src/main/scala/com/salesforce/op/filters/FeatureDistribution.scala

@@ -32,8 +32,8 @@ package com.salesforce.op.filters
 
 import java.util.Objects
 
-import com.salesforce.op.features.{FeatureDistributionLike, FeatureDistributionType}
-import com.salesforce.op.stages.impl.feature.{HashAlgorithm, Inclusion, NumericBucketizer, TextStats}
+import com.salesforce.op.features.{FeatureDistributionLike, FeatureDistributionType, TextStats}
+import com.salesforce.op.stages.impl.feature.{HashAlgorithm, Inclusion, NumericBucketizer}
 import com.salesforce.op.utils.json.EnumEntrySerializer
 import com.twitter.algebird.Monoid._
 import com.twitter.algebird._
@@ -63,7 +63,6 @@ case class FeatureDistribution
  nulls: Long,
  distribution: Array[Double],
  summaryInfo: Array[Double],
- moments: Option[Moments] = None,
  cardEstimate: Option[TextStats] = None,
  `type`: FeatureDistributionType = FeatureDistributionType.Training
 ) extends FeatureDistributionLike {
@@ -109,11 +108,23 @@ case class FeatureDistribution
  // summary info can be empty or min max if hist is empty but should otherwise match so take the longest info
  val combinedSummaryInfo = if (summaryInfo.length > fd.summaryInfo.length) summaryInfo else fd.summaryInfo
 
- val combinedMoments = moments + fd.moments
  val combinedCard = cardEstimate + fd.cardEstimate
 
  FeatureDistribution(name, key, count + fd.count, nulls + fd.nulls, combinedDist,
- combinedSummaryInfo, combinedMoments, combinedCard, `type`)
+ combinedSummaryInfo, combinedCard, `type`)
+ }
+
+ /**
+ * Cardinality of the length of tokenized text, or numerical value for numbers (at most 500)
+ *
+ * @return cardinality count based on cardEstimate
+ */
+
+ def cardSize(): Option[Int] = {
+ cardEstimate match {
+ case Some(x) => Option(x.valueCounts.size)
+ case _ => None
+ }
  }
 
  /**
@@ -166,22 +177,20 @@ case class FeatureDistribution
  "count" -> count.toString,
  "nulls" -> nulls.toString,
  "distribution" -> distribution.mkString("[", ",", "]"),
- "summaryInfo" -> summaryInfo.mkString("[", ",", "]"),
- "moments" -> moments.map(_.toString).getOrElse("")
+ "summaryInfo" -> summaryInfo.mkString("[", ",", "]")
  ).map { case (n, v) => s"$n = $v" }.mkString(", ")
 
  s"${getClass.getSimpleName}($valStr)"
  }
 
  override def equals(that: Any): Boolean = that match {
- case FeatureDistribution(`name`, `key`, `count`, `nulls`, d, s, m, c, `type`) =>
- distribution.deep == d.deep && summaryInfo.deep == s.deep &&
- moments == m && cardEstimate == c
+ case FeatureDistribution(`name`, `key`, `count`, `nulls`, d, s, c, `type`) =>
+ distribution.deep == d.deep && summaryInfo.deep == s.deep && cardEstimate == c
  case _ => false
  }
 
  override def hashCode(): Int = Objects.hashCode(name, key, count, nulls, distribution,
- summaryInfo, moments, cardEstimate, `type`)
+ summaryInfo, cardEstimate, `type`)
 }
 
 object FeatureDistribution {
@@ -239,7 +248,6 @@ object FeatureDistribution {
  value.map(seq => 0L -> histValues(seq, summary, bins, textBinsFormula))
  .getOrElse(1L -> (Array(summary.min, summary.max, summary.sum, summary.count) -> new Array[Double](bins)))
 
- val moments = value.map(momentsValues)
  val cardEstimate = value.map(cardinalityValues)
 
  FeatureDistribution(
@@ -249,26 +257,11 @@ object FeatureDistribution {
  nulls = nullCount,
  summaryInfo = summaryInfo,
  distribution = distribution,
- moments = moments,
  cardEstimate = cardEstimate,
  `type` = `type`
  )
  }
 
- /**
- * Function to calculate the first five central moments of numeric values, or length of tokens for text features
- *
- * @param values values to calculate moments
- * @return Moments object containing information about moments
- */
- private def momentsValues(values: ProcessedSeq): Moments = {
- val population = values match {
- case Left(seq) => seq.map(x => x.length.toDouble)
- case Right(seq) => seq
- }
- MomentsGroup.sum(population.map(x => Moments(x)))
- }
-
  /**
  * Function to track frequency of the first $(MaxCardinality) unique values
  * (number for numeric features, token for text features)

core/src/main/scala/com/salesforce/op/filters/RawFeatureFilter.scala

@@ -216,7 +216,7 @@ class RawFeatureFilter[T]
  val featureSize: Int = trainingDistribs.length
 
  val trainingFillRates: Seq[Double] = trainingDistribs.map(_.fillRate())
-
+ val trainingCardSizes: Seq[Option[Int]] = trainingDistribs.map(_.cardSize)
  val trainingNullLabelAbsoluteCorrs: Seq[Option[Double]] =
  if (correlationInfo.isEmpty) Seq.fill(featureSize)(None)
  else {
@@ -232,6 +232,7 @@ class RawFeatureFilter[T]
  traininingDistribs: Seq[FeatureDistribution],
  trainingFillRates: Seq[Double],
  trainingNullLabelAbsoluteCorrs: Seq[Option[Double]],
+ trainingCardSizes: Seq[Option[Int]],
  scoringFillRates: Seq[Option[Double]],
  jsDivergences: Seq[Option[Double]],
  fillRateDiffs: Seq[Option[Double]],
@@ -245,12 +246,15 @@ class RawFeatureFilter[T]
  .zip(jsDivergences)
  .zip(fillRateDiffs)
  .zip(fillRatioDiffs)
+ .zip(trainingCardSizes)
  .map {
- case (((((((name, key), trainingFillRate), trainingNullLabelAbsoluteCorr),
- scoringFillRate), jsDivergence), fillRateDiff), fillRatioDiff) =>
+ case ((((((((name, key), trainingFillRate), trainingNullLabelAbsoluteCorr),
+ scoringFillRate), jsDivergence), fillRateDiff), fillRatioDiff),
+ trainingCardSize) =>
  RawFeatureFilterMetrics(
  name, key, trainingFillRate, trainingNullLabelAbsoluteCorr,
- scoringFillRate, jsDivergence, fillRateDiff, fillRatioDiff)
+ scoringFillRate, jsDivergence, fillRateDiff, fillRatioDiff,
+ trainingCardSize)
  }
  }
 
@@ -263,7 +267,7 @@ class RawFeatureFilter[T]
 
  val rawFeatureFilterMetrics = combineRawFeatureFilterMetrics(
  trainingDistribs, trainingFillRates, trainingNullLabelAbsoluteCorrs,
- scoringFillRates, jsDivergences, fillRateDiffs, fillRatioDiffs
+ trainingCardSizes, scoringFillRates, jsDivergences, fillRateDiffs, fillRatioDiffs
  )
  rawFeatureFilterMetrics
 
@@ -280,7 +284,8 @@ class RawFeatureFilter[T]
 
  val rawFeatureFilterMetrics = combineRawFeatureFilterMetrics(
  trainingDistribs, trainingFillRates, trainingNullLabelAbsoluteCorrs,
- scoringFillRates, jsDivergences, fillRateDiffs, fillRatioDiffs
+ trainingCardSizes, scoringFillRates, jsDivergences,
+ fillRateDiffs, fillRatioDiffs
  )
 
  log.info(combined.zip(rawFeatureFilterMetrics).map {
@@ -354,7 +359,8 @@ class RawFeatureFilter[T]
  .zip(fillRatioDiffMismatches)
  .map {
  case (((((((name, key), trainingUnfilledState), trainingNullLabelLeaker),
- scoringUnfilledState), jsDivergenceMismatch), fillRateDiffMismatch), fillRatioDiffMismatch) =>
+ scoringUnfilledState), jsDivergenceMismatch),
+ fillRateDiffMismatch), fillRatioDiffMismatch) =>
  ExclusionReasons(
  name,
  key,

core/src/main/scala/com/salesforce/op/filters/RawFeatureFilterResults.scala

@@ -142,7 +142,8 @@ case class RawFeatureFilterMetrics
  scoringFillRate: Option[Double],
  jsDivergence: Option[Double],
  fillRateDiff: Option[Double],
- fillRatioDiff: Option[Double]
+ fillRatioDiff: Option[Double],
+ trainingCardSize: Option[Int]
 )
 
 /**

core/src/main/scala/com/salesforce/op/stages/impl/feature/IdMapRemover.scala

@@ -0,0 +1,32 @@
+package com.salesforce.op.stages.impl.feature
+
+import com.salesforce.op.UID
+import com.salesforce.op.features.types.{TextMap}
+import com.salesforce.op.stages.base.unary.UnaryTransformer
+
+class IdMapRemover(
+ minUniqueTokLen: Int,
+ uid: String = UID[IdMapRemover]
+) extends UnaryTransformer[TextMap, TextMap](operationName = "IdMapRemover", uid = uid) {
+
+ private var dropMap: Map[String, Boolean] = Map()
+
+ override protected def onSetInput(): Unit = {
+ super.onSetInput()
+ val dist = in1.asFeatureLike.distributions
+ val keys = dist.flatMap(_.key)
+ val drop = dist.flatMap(_.cardEstimate).map(_.valueCounts.size < minUniqueTokLen)
+ dropMap = (keys zip drop) toMap
+ }
+
+ override def transformFn: TextMap => TextMap =
+ a => {
+ val filteredMap = a.value.map { case (k, v) =>
+ dropMap.get(k) match {
+ case Some(true) => (k, "")
+ case _ => (k, v)
+ }
+ }
+ TextMap(filteredMap)
+ }
+}

core/src/main/scala/com/salesforce/op/stages/impl/feature/IdRemover.scala

@@ -0,0 +1,23 @@
+package com.salesforce.op.stages.impl.feature
+
+import com.salesforce.op.UID
+import com.salesforce.op.features.types.Text
+import com.salesforce.op.stages.base.unary.UnaryTransformer
+
+class IdRemover(
+ minUniqueTokLen: Int,
+ uid: String = UID[IdRemover],
+ operationName: String = "IDremover"
+) extends UnaryTransformer[Text, Text] (operationName = operationName, uid = uid) {
+
+ private var drop: Boolean = false
+
+ override protected def onSetInput(): Unit = {
+ super.onSetInput()
+ val dist = in1.asFeatureLike.distributions
+ val tokenLenCardFilter = dist.flatMap(_.cardEstimate).map(_.valueCounts.size < minUniqueTokLen)
+ drop = tokenLenCardFilter.headOption.getOrElse(false)
+ }
+
+ override def transformFn: Text => Text = a => if (drop) Text.empty else a
+}

core/src/main/scala/com/salesforce/op/stages/impl/feature/SmartTextMapVectorizer.scala

@@ -31,6 +31,7 @@
 package com.salesforce.op.stages.impl.feature
 
 import com.salesforce.op.UID
+import com.salesforce.op.features.TextStats
 import com.salesforce.op.features.types._
 import com.salesforce.op.stages.base.sequence.{SequenceEstimator, SequenceModel}
 import com.salesforce.op.stages.impl.feature.VectorizerUtils._

core/src/main/scala/com/salesforce/op/stages/impl/feature/SmartTextVectorizer.scala

@@ -31,14 +31,13 @@
 package com.salesforce.op.stages.impl.feature
 
 import com.salesforce.op.UID
-import com.salesforce.op.features.TransientFeature
-import com.salesforce.op.features.types.{OPVector, Text, TextList, VectorConversions, SeqDoubleConversions}
+import com.salesforce.op.features.{TextStats, TransientFeature}
+import com.salesforce.op.features.types.{OPVector, SeqDoubleConversions, Text, TextList, VectorConversions}
 import com.salesforce.op.stages.base.sequence.{SequenceEstimator, SequenceModel}
 import com.salesforce.op.stages.impl.feature.VectorizerUtils._
 import com.salesforce.op.utils.json.JsonLike
 import com.salesforce.op.utils.spark.RichDataset._
 import com.salesforce.op.utils.spark.{OpVectorColumnMetadata, OpVectorMetadata}
-import com.twitter.algebird.Monoid
 import com.twitter.algebird.Monoid._
 import com.twitter.algebird.Operators._
 import com.twitter.algebird.Semigroup
@@ -163,27 +162,6 @@ object SmartTextVectorizer {
  }
 }
 
-/**
- * Summary statistics of a text feature
- *
- * @param valueCounts counts of feature values
- */
-private[op] case class TextStats(valueCounts: Map[String, Int]) extends JsonLike
-
-private[op] object TextStats {
- def monoid(maxCardinality: Int): Monoid[TextStats] = new Monoid[TextStats] {
- override def plus(l: TextStats, r: TextStats): TextStats = {
- if (l.valueCounts.size > maxCardinality) l
- else if (r.valueCounts.size > maxCardinality) r
- else TextStats(l.valueCounts + r.valueCounts)
- }
-
- override def zero: TextStats = TextStats.empty
- }
-
- def empty: TextStats = TextStats(Map.empty)
-}
-
 /**
  * Arguments for [[SmartTextVectorizerModel]]
  *

core/src/test/scala/com/salesforce/op/ModelInsightsTest.scala

@@ -32,13 +32,13 @@ package com.salesforce.op
 
 import com.salesforce.op.evaluators._
 import com.salesforce.op.features.types._
-import com.salesforce.op.features.{Feature, FeatureDistributionType, FeatureLike}
+import com.salesforce.op.features.{Feature, FeatureDistributionType, FeatureLike, TextStats}
 import com.salesforce.op.filters._
 import com.salesforce.op.stages.impl.classification._
 import com.salesforce.op.stages.impl.preparators._
 import com.salesforce.op.stages.impl.regression.{OpLinearRegression, OpXGBoostRegressor, RegressionModelSelector}
 import com.salesforce.op.stages.impl.selector.ModelSelectorNames.EstimatorType
-import com.salesforce.op.stages.impl.selector.{SelectedModelCombiner, SelectedCombinerModel, SelectedModel}
+import com.salesforce.op.stages.impl.selector.{SelectedCombinerModel, SelectedModel, SelectedModelCombiner}
 import com.salesforce.op.stages.impl.selector.ValidationType._
 import com.salesforce.op.stages.impl.tuning.{DataCutter, DataSplitter}
 import com.salesforce.op.test.{PassengerSparkFixtureTest, TestFeatureBuilder}
@@ -48,8 +48,7 @@ import org.apache.spark.ml.param.ParamMap
 import org.apache.spark.ml.tuning.ParamGridBuilder
 import org.junit.runner.RunWith
 import com.salesforce.op.features.types.Real
-import com.salesforce.op.stages.impl.feature.{CombinationStrategy, TextStats}
-import com.twitter.algebird.Moments
+import com.salesforce.op.stages.impl.feature.CombinationStrategy
 import org.apache.spark.sql.{DataFrame, Dataset}
 import org.scalactic.Equality
 import org.scalatest.FlatSpec
@@ -167,15 +166,14 @@ class ModelInsightsTest extends FlatSpec with PassengerSparkFixtureTest with Dou
  }
 
  def getFeatureMomentsAndCard(inputModel: FeatureLike[Prediction],
- DF: DataFrame): (Map[String, Moments], Map[String, TextStats]) = {
+ DF: DataFrame): Map[String, TextStats] = {
  lazy val workFlow = new OpWorkflow().setResultFeatures(inputModel).setInputDataset(DF)
  lazy val dummyReader = workFlow.getReader()
  lazy val workFlowRFF = workFlow.withRawFeatureFilter(Some(dummyReader), None)
  lazy val model = workFlowRFF.train()
  val insights = model.modelInsights(inputModel)
- val featureMoments = insights.features.map(f => f.featureName -> f.distributions.head.moments.get).toMap
  val featureCardinality = insights.features.map(f => f.featureName -> f.distributions.head.cardEstimate.get).toMap
- return (featureMoments, featureCardinality)
+ return featureCardinality
  }
 
  val params = new OpParams()
@@ -777,23 +775,15 @@ class ModelInsightsTest extends FlatSpec with PassengerSparkFixtureTest with Dou
  absError2 should be < tol * smallCoeffSum / 2
  }
 
- it should "correctly return moments calculation and cardinality calculation for numeric features" in {
+ it should "correctly return cardinality calculation for numeric features" in {
 
  import spark.implicits._
  val df = linRegDF._3
  val meanTol = 0.01
  val varTol = 0.01
- val (moments, cardinality) = getFeatureMomentsAndCard(standardizedLinpred, linRegDF._3)
+ val cardinality = getFeatureMomentsAndCard(standardizedLinpred, linRegDF._3)
 
  // Go through each feature and check that the mean, variance, and unique counts match the data
- moments.foreach { case (featureName, value) => {
- value.count shouldBe 1000
- val (expectedMean, expectedVariance) =
- df.select(avg(featureName), variance(featureName)).as[(Double, Double)].collect().head
- math.abs((value.mean - expectedMean) / expectedMean) < meanTol shouldBe true
- math.abs((value.variance - expectedVariance) / expectedVariance) < varTol shouldBe true
- }
- }
 
  cardinality.foreach { case (featureName, value) => {
  val actualUniques = df.select(featureName).as[Double].collect().toSet