Skip to content

[SPARK-22397][ML]add multiple columns support to QuantileDiscretizer #19715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
huaxingao wants to merge 9 commits into from
Closed

[SPARK-22397][ML]add multiple columns support to QuantileDiscretizer #19715

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
07bd868
[SPARK-22397][ML]add multiple columns support to QuantileDiscretizer
huaxingao Nov 10, 2017
87ee0f3
fix binary compatibility issue
huaxingao Nov 11, 2017
5038e21
address comments
huaxingao Nov 19, 2017
97ad483
address comments
huaxingao Nov 30, 2017
445bd84
address comments
huaxingao Dec 9, 2017
0e5971b
throw Exception if both inputCol and inputCols are set
huaxingao Dec 12, 2017
a030da1
Address Comments
huaxingao Dec 12, 2017
99726a1
Address Comments for test case
huaxingao Dec 15, 2017
486b68d
address comment
huaxingao Dec 21, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 104 additions & 16 deletions mllib/src/main/scala/org/apache/spark/ml/feature/QuantileDiscretizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.ml._
import org.apache.spark.ml.attribute.NominalAttribute
import org.apache.spark.ml.param._
import org.apache.spark.ml.param.shared.{HasHandleInvalid, HasInputCol, HasOutputCol}
import org.apache.spark.ml.param.shared.{HasHandleInvalid, HasInputCol, HasInputCols, HasOutputCol, HasOutputCols}
import org.apache.spark.ml.util._
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.types.StructType
Expand Down Expand Up @@ -50,10 +50,28 @@ private[feature] trait QuantileDiscretizerBase extends Params
/** @group getParam */
def getNumBuckets: Int = getOrDefault(numBuckets)

/**
* Array of number of buckets (quantiles, or categories) into which data points are grouped.

@MLnick MLnick Nov 29, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can add a comment about "each value must be greater than or equal to 2"

* Each value must be greater than or equal to 2
*
* See also [[handleInvalid]], which can optionally create an additional bucket for NaN values.
*
* @group param
*/
val numBucketsArray = new IntArrayParam(this, "numBucketsArray", "Array of number of buckets " +
"(quantiles, or categories) into which data points are grouped. This is for multiple " +
"columns input. If transforming multiple columns and numBucketsArray is not set, but " +
"numBuckets is set, then numBuckets will be applied across all columns.",
(arrayOfNumBuckets: Array[Int]) => arrayOfNumBuckets.forall(ParamValidators.gtEq(2)))

/** @group getParam */
def getNumBucketsArray: Array[Int] = $(numBucketsArray)

/**
* Relative error (see documentation for
* `org.apache.spark.sql.DataFrameStatFunctions.approxQuantile` for description)
* Must be in the range [0, 1].
* Note that in multiple columns case, relative error is applied to all columns.
* default: 0.001
* @group param
*/
Expand All @@ -68,7 +86,9 @@ private[feature] trait QuantileDiscretizerBase extends Params
/**
* Param for how to handle invalid entries. Options are 'skip' (filter out rows with
* invalid values), 'error' (throw an error), or 'keep' (keep invalid values in a special
* additional bucket).
* additional bucket). Note that in the multiple columns case, the invalid handling is applied
* to all columns. That said for 'error' it will throw an error if any invalids are found in
* any column, for 'skip' it will skip rows with any invalids in any columns, etc.
* Default: "error"
* @group param
*/
Expand All @@ -86,6 +106,11 @@ private[feature] trait QuantileDiscretizerBase extends Params
* categorical features. The number of bins can be set using the `numBuckets` parameter. It is
* possible that the number of buckets used will be smaller than this value, for example, if there
* are too few distinct values of the input to create enough distinct quantiles.
* Since 2.3.0, `QuantileDiscretizer` can map multiple columns at once by setting the `inputCols`
* parameter. If both of the `inputCol` and `inputCols` parameters are set, an Exception will be
* thrown. To specify the number of buckets for each column, the `numBucketsArray` parameter can
* be set, or if the number of buckets should be the same across columns, `numBuckets` can be
* set as a convenience.
*
* NaN handling:
* null and NaN values will be ignored from the column during `QuantileDiscretizer` fitting. This
Expand All @@ -104,7 +129,8 @@ private[feature] trait QuantileDiscretizerBase extends Params
*/
@Since("1.6.0")
final class QuantileDiscretizer @Since("1.6.0") (@Since("1.6.0") override val uid: String)
extends Estimator[Bucketizer] with QuantileDiscretizerBase with DefaultParamsWritable {
extends Estimator[Bucketizer] with QuantileDiscretizerBase with DefaultParamsWritable
with HasInputCols with HasOutputCols {

@viirya viirya Nov 12, 2017

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks a bit weird to have HasInputCols and HasOutputCols directly in QuantileDiscretizer and leave other params in QuantileDiscretizerBase.

But extending HasInputCols and HasOutputCols in QuantileDiscretizerBase causes binary compatibility issue. I think we don't want to break the compatibility in the effort of adding multi-col support.

@huaxingao huaxingao Nov 13, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I will leave this as is even though it's a bit weird.


@Since("1.6.0")
def this() = this(Identifiable.randomUID("quantileDiscretizer"))
Expand All @@ -129,34 +155,96 @@ final class QuantileDiscretizer @Since("1.6.0") (@Since("1.6.0") override val ui
@Since("2.1.0")
def setHandleInvalid(value: String): this.type = set(handleInvalid, value)

/** @group setParam */
@Since("2.3.0")
def setNumBucketsArray(value: Array[Int]): this.type = set(numBucketsArray, value)

/** @group setParam */
@Since("2.3.0")
def setInputCols(value: Array[String]): this.type = set(inputCols, value)

/** @group setParam */
@Since("2.3.0")
def setOutputCols(value: Array[String]): this.type = set(outputCols, value)

private[feature] def getInOutCols: (Array[String], Array[String]) = {
require((isSet(inputCol) && isSet(outputCol) && !isSet(inputCols) && !isSet(outputCols)) ||
(!isSet(inputCol) && !isSet(outputCol) && isSet(inputCols) && isSet(outputCols)),
"QuantileDiscretizer only supports setting either inputCol/outputCol or" +
"inputCols/outputCols."
)

if (isSet(inputCol)) {
(Array($(inputCol)), Array($(outputCol)))
} else {
require($(inputCols).length == $(outputCols).length,

@MLnick MLnick Dec 15, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should add a small test case for mismatched sizes of inputCols / outputCols.

"inputCols number do not match outputCols")
($(inputCols), $(outputCols))
}
}

@Since("1.6.0")
override def transformSchema(schema: StructType): StructType = {
SchemaUtils.checkNumericType(schema, $(inputCol))
val inputFields = schema.fields
require(inputFields.forall(_.name != $(outputCol)),
s"Output column ${$(outputCol)} already exists.")
val attr = NominalAttribute.defaultAttr.withName($(outputCol))
val outputFields = inputFields :+ attr.toStructField()
val (inputColNames, outputColNames) = getInOutCols
val existingFields = schema.fields
var outputFields = existingFields
inputColNames.zip(outputColNames).foreach { case (inputColName, outputColName) =>
SchemaUtils.checkNumericType(schema, inputColName)
require(existingFields.forall(_.name != outputColName),
s"Output column ${outputColName} already exists.")
val attr = NominalAttribute.defaultAttr.withName(outputColName)
outputFields :+= attr.toStructField()
}
StructType(outputFields)
}

@Since("2.0.0")
override def fit(dataset: Dataset[_]): Bucketizer = {
transformSchema(dataset.schema, logging = true)
val splits = dataset.stat.approxQuantile($(inputCol),
(0.0 to 1.0 by 1.0/$(numBuckets)).toArray, $(relativeError))
val bucketizer = new Bucketizer(uid).setHandleInvalid($(handleInvalid))

@MLnick MLnick Dec 8, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking at this now, the Array.fill approach probably adds needless complexity.

But the multi-buckets case can perhaps still be cleaned up. How about something like this:

  override def fit(dataset: Dataset[_]): Bucketizer = {
    transformSchema(dataset.schema, logging = true)
    val bucketizer = new Bucketizer(uid).setHandleInvalid($(handleInvalid))
    if (isQuantileDiscretizeMultipleColumns) {
      val splitsArray = if (isSet(numBucketsArray)) {
        val probArrayPerCol = $(numBucketsArray).map { numOfBuckets =>
          (0.0 to 1.0 by 1.0 / numOfBuckets).toArray
        }

        val probabilityArray = probArrayPerCol.flatten.sorted.distinct
        val splitsArrayRaw = dataset.stat.approxQuantile($(inputCols),
          probabilityArray, $(relativeError))

        splitsArrayRaw.zip(probArrayPerCol).map { case (splits, probs) =>
          val probSet = probs.toSet
          val idxSet = probabilityArray.zipWithIndex.collect {
            case (p, idx) if probSet(p) =>
              idx
          }.toSet
          splits.zipWithIndex.collect {
            case (s, idx) if idxSet(idx) =>
              s
          }
        }
      } else {
        dataset.stat.approxQuantile($(inputCols),
          (0.0 to 1.0 by 1.0 / $(numBuckets)).toArray, $(relativeError))
      }
      bucketizer.setSplitsArray(splitsArray.map(getDistinctSplits))
    } else {
      val splits = dataset.stat.approxQuantile($(inputCol),
        (0.0 to 1.0 by 1.0 / $(numBuckets)).toArray, $(relativeError))
      bucketizer.setSplits(getDistinctSplits(splits))
    }
    copyValues(bucketizer.setParent(this))
  }

Then we don't need getSplitsForEachColumn method (or part of the above could be factored out into a private method if it makes sense).

if (isSet(inputCols)) {
val splitsArray = if (isSet(numBucketsArray)) {
val probArrayPerCol = $(numBucketsArray).map { numOfBuckets =>
(0.0 to 1.0 by 1.0 / numOfBuckets).toArray
}

val probabilityArray = probArrayPerCol.flatten.sorted.distinct
val splitsArrayRaw = dataset.stat.approxQuantile($(inputCols),
probabilityArray, $(relativeError))

splitsArrayRaw.zip(probArrayPerCol).map { case (splits, probs) =>
val probSet = probs.toSet
val idxSet = probabilityArray.zipWithIndex.collect {
case (p, idx) if probSet(p) =>
idx
}.toSet
splits.zipWithIndex.collect {
case (s, idx) if idxSet(idx) =>
s
}
}
} else {
dataset.stat.approxQuantile($(inputCols),
(0.0 to 1.0 by 1.0 / $(numBuckets)).toArray, $(relativeError))
}
bucketizer.setSplitsArray(splitsArray.map(getDistinctSplits))
} else {
val splits = dataset.stat.approxQuantile($(inputCol),
(0.0 to 1.0 by 1.0 / $(numBuckets)).toArray, $(relativeError))
bucketizer.setSplits(getDistinctSplits(splits))
}
copyValues(bucketizer.setParent(this))
}

private def getDistinctSplits(splits: Array[Double]): Array[Double] = {
splits(0) = Double.NegativeInfinity
splits(splits.length - 1) = Double.PositiveInfinity

val distinctSplits = splits.distinct
if (splits.length != distinctSplits.length) {
log.warn(s"Some quantiles were identical. Bucketing to ${distinctSplits.length - 1}" +
s" buckets as a result.")
}
val bucketizer = new Bucketizer(uid)
.setSplits(distinctSplits.sorted)
.setHandleInvalid($(handleInvalid))
copyValues(bucketizer.setParent(this))
distinctSplits.sorted
}

@Since("1.6.0")
Expand Down
Loading