adjust bin value based on number of token

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

@@ -31,7 +31,7 @@
 package com.salesforce.op.filters
 
 import com.salesforce.op.features.TransientFeature
-import com.salesforce.op.stages.impl.feature.{Inclusion, NumericBucketizer}
+import com.salesforce.op.stages.impl.feature.{HashAlgorithm, Inclusion, NumericBucketizer}
 import com.twitter.algebird.Semigroup
 import com.twitter.algebird.Monoid._
 import com.twitter.algebird.Operators._
@@ -142,6 +142,7 @@ case class FeatureDistribution
 private[op] object FeatureDistribution {
 
  val MaxBins = 100000
+ val AvgBinValue = 5000
 
  implicit val semigroup: Semigroup[FeatureDistribution] = new Semigroup[FeatureDistribution] {
  override def plus(l: FeatureDistribution, r: FeatureDistribution) = l.reduce(r)
@@ -154,18 +155,16 @@ private[op] object FeatureDistribution {
  * @param summary feature summary
  * @param value optional processed sequence
  * @param bins number of histogram bins
- * @param hasher hashing method to use for text and categorical features
  * @return feature distribution given the provided information
  */
  def apply(
  featureKey: FeatureKey,
  summary: Summary,
  value: Option[ProcessedSeq],
- bins: Int,
- hasher: HashingTF
+ bins: Int
  ): FeatureDistribution = {
  val (nullCount, (summaryInfo, distribution)): (Int, (Array[Double], Array[Double])) =
- value.map(seq => 0 -> histValues(seq, summary, bins, hasher))
+ value.map(seq => 0 -> histValues(seq, summary, bins))
  .getOrElse(1 -> (Array(summary.min, summary.max) -> Array.fill(bins)(0.0)))
 
  FeatureDistribution(
@@ -182,18 +181,25 @@ private[op] object FeatureDistribution {
  * @param values values to bin
  * @param sum summary info for feature (max and min)
  * @param bins number of bins to produce
- * @param hasher hasing function to use for text
  * @return the bin information and the binned counts
  */
  // TODO avoid wrapping and unwrapping??
  private def histValues(
  values: ProcessedSeq,
  sum: Summary,
- bins: Int,
- hasher: HashingTF
+ bins: Int
  ): (Array[Double], Array[Double]) = {
  values match {
- case Left(seq) => Array(sum.min, sum.max) -> hasher.transform(seq).toArray // TODO use summary info to pick hashes
+ case Left(seq) => {
+ val minBins = bins
+ val maxBins = MaxBins
+ val numBins = math.min(math.max(bins, sum.max/AvgBinValue), maxBins).toInt
+
+ val hasher: HashingTF = new HashingTF(numFeatures = numBins)
+ .setBinary(false)
+ .setHashAlgorithm(HashAlgorithm.MurMur3.toString.toLowerCase)
+ Array(sum.min, sum.max) -> hasher.transform(seq).toArray
+ }
  case Right(seq) => // TODO use kernel fit instead of histogram
  if (sum == Summary.empty) {
  Array(sum.min, sum.max) -> seq.toArray // the seq will always be empty in this case

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

@@ -88,29 +88,26 @@ private[filters] case class PreparedFeatures(
  def getFeatureDistributions(
  responseSummaries: Array[(FeatureKey, Summary)],
  predictorSummaries: Array[(FeatureKey, Summary)],
- bins: Int,
- hasher: HashingTF
+ bins: Int
  ): (Array[FeatureDistribution], Array[FeatureDistribution]) = {
  val responseFeatureDistributions: Array[FeatureDistribution] =
- getFeatureDistributions(responses, responseSummaries, bins, hasher)
+ getFeatureDistributions(responses, responseSummaries, bins)
  val predictorFeatureDistributions: Array[FeatureDistribution] =
- getFeatureDistributions(predictors, predictorSummaries, bins, hasher)
+ getFeatureDistributions(predictors, predictorSummaries, bins)
 
  responseFeatureDistributions -> predictorFeatureDistributions
  }
 
  private def getFeatureDistributions(
  features: Map[FeatureKey, ProcessedSeq],
  summaries: Array[(FeatureKey, Summary)],
- bins: Int,
- hasher: HashingTF
+ bins: Int
  ): Array[FeatureDistribution] = summaries.map { case (featureKey, summary) =>
  FeatureDistribution(
  featureKey = featureKey,
  summary = summary,
  value = features.get(featureKey),
- bins = bins,
- hasher = hasher)
+ bins = bins)
  }
 }
 

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

@@ -95,10 +95,6 @@ class RawFeatureFilter[T]
 
  @transient protected lazy val log = LoggerFactory.getLogger(this.getClass)
 
- private val hasher: HashingTF = new HashingTF(numFeatures = bins)
- .setBinary(false)
- .setHashAlgorithm(HashAlgorithm.MurMur3.toString.toLowerCase)
-
 
  /**
  * Get binned counts of the feature distribution and empty count for each raw feature
@@ -138,8 +134,7 @@ class RawFeatureFilter[T]
  .map(_.getFeatureDistributions(
  responseSummaries = responseSummariesArr,
  predictorSummaries = predictorSummariesArr,
- bins = bins,
- hasher = hasher))
+ bins = bins))
  .reduce(_ + _) // NOTE: resolved semigroup is IndexedSeqSemigroup
  val correlationInfo: Map[FeatureKey, Map[FeatureKey, Double]] =
  allFeatureInfo.map(_.correlationInfo).getOrElse {

core/src/test/scala/com/salesforce/op/filters/FeatureDistributionTest.scala

@@ -30,11 +30,10 @@
 
 package com.salesforce.op.filters
 
-import com.salesforce.op.OpParams
-import com.salesforce.op.features.{OPFeature, TransientFeature}
+import com.salesforce.op.features.TransientFeature
 import com.salesforce.op.stages.impl.feature.HashAlgorithm
 import com.salesforce.op.test.PassengerSparkFixtureTest
-import com.salesforce.op.utils.spark.RichDataset._
+import com.salesforce.op.testkit.RandomText
 import org.apache.spark.mllib.feature.HashingTF
 import org.junit.runner.RunWith
 import org.scalatest.FlatSpec
@@ -52,16 +51,13 @@ class FeatureDistributionTest extends FlatSpec with PassengerSparkFixtureTest wi
  val summary =
  Array(Summary(0.0, 1.0), Summary(-1.6, 10.6), Summary(0.0, 3.0), Summary(0.0, 0.0), Summary(1.0, 5.0))
  val bins = 10
- val hasher: HashingTF = new HashingTF(numFeatures = bins)
- .setBinary(false)
- .setHashAlgorithm(HashAlgorithm.MurMur3.toString.toLowerCase)
 
  val featureKeys: Array[FeatureKey] = features.map(f => (f.name, None))
  val processedSeqs: Array[Option[ProcessedSeq]] = values.map { case (isEmpty, processed) =>
  if (isEmpty) None else Option(processed)
  }
  val distribs = featureKeys.zip(summary).zip(processedSeqs).map { case ((key, summ), seq) =>
- FeatureDistribution(key, summ, seq, bins, hasher)
+ FeatureDistribution(key, summ, seq, bins)
  }
  distribs.foreach{ d =>
  d.key shouldBe None
@@ -78,6 +74,29 @@ class FeatureDistributionTest extends FlatSpec with PassengerSparkFixtureTest wi
  distribs(4).summaryInfo.length shouldBe bins
  }
 
+ it should "be correctly created for text features" in {
+ val features = Array(description, gender)
+ val values: Array[(Boolean, ProcessedSeq)] = Array(
+ (false, Left(RandomText.strings(1, 10).take(10000).toSeq.map(_.value.get))),
+ (false, Left(RandomText.strings(1, 10).take(1000000).toSeq.map(_.value.get)))
+ )
+ val summary = Array(Summary(10000.0, 10000.0), Summary(1000000, 1000000))
+ val bins = 100
+ val featureKeys: Array[FeatureKey] = features.map(f => (f.name, None))
+ val processedSeqs: Array[Option[ProcessedSeq]] = values.map { case (isEmpty, processed) =>
+ if (isEmpty) None else Option(processed)
+ }
+ val distribs = featureKeys.zip(summary).zip(processedSeqs).map { case ((key, summ), seq) =>
+ FeatureDistribution(key, summ, seq, bins)
+ }
+
+ distribs(0).distribution.length shouldBe 100
+ distribs(0).distribution.sum shouldBe 10000
+
+ distribs(1).distribution.length shouldBe 200
+ distribs(1).distribution.sum shouldBe 1000000
+ }
+
  it should "be correctly created for map features" in {
  val features = Array(stringMap, numericMap, booleanMap).map(TransientFeature.apply)
  val values: Array[Map[String, ProcessedSeq]] = Array(
@@ -95,7 +114,7 @@ class FeatureDistributionTest extends FlatSpec with PassengerSparkFixtureTest wi
  val distribs = features.map(_.name).zip(summary).zip(values).flatMap { case ((name, summaryMaps), valueMaps) =>
  summaryMaps.map { case (key, summary) =>
  val featureKey = (name, Option(key))
- FeatureDistribution(featureKey, summary, valueMaps.get(key), bins, hasher)
+ FeatureDistribution(featureKey, summary, valueMaps.get(key), bins)
  }
  }