diff --git a/docs/ml-features.md b/docs/ml-features.md index 2da13576c4ef4..92d2f3d0b418d 100644 --- a/docs/ml-features.md +++ b/docs/ml-features.md @@ -1009,6 +1009,51 @@ for more details on the API. + +## RobustScaler + +`RobustScaler` transforms a dataset of `Vector` rows, removing the median and scaling the data according to a specific quantile range (by default the IQR: Interquartile Range, quantile range between the 1st quartile and the 3rd quartile). Its behavior is quite similar to `StandardScaler`, however the median and the quantile range are used instead of mean and standard deviation, which make it robust to outliers. It takes parameters: + +* `lower`: 0.25 by default. Lower quantile to calculate quantile range, shared by all features. +* `upper`: 0.75 by default. Upper quantile to calculate quantile range, shared by all features. +* `withScaling`: True by default. Scales the data to quantile range. +* `withCentering`: False by default. Centers the data with median before scaling. It will build a dense output, so take care when applying to sparse input. + +`RobustScaler` is an `Estimator` which can be `fit` on a dataset to produce a `RobustScalerModel`; this amounts to computing quantile statistics. The model can then transform a `Vector` column in a dataset to have unit quantile range and/or zero median features. + +Note that if the quantile range of a feature is zero, it will return default `0.0` value in the `Vector` for that feature. + +**Examples** + +The following example demonstrates how to load a dataset in libsvm format and then normalize each feature to have unit quantile range. + +
+
+ +Refer to the [RobustScaler Scala docs](api/scala/index.html#org.apache.spark.ml.feature.RobustScaler) +for more details on the API. + +{% include_example scala/org/apache/spark/examples/ml/RobustScalerExample.scala %} +
+ +
+ +Refer to the [RobustScaler Java docs](api/java/org/apache/spark/ml/feature/RobustScaler.html) +for more details on the API. + +{% include_example java/org/apache/spark/examples/ml/JavaRobustScalerExample.java %} +
+ +
+ +Refer to the [RobustScaler Python docs](api/python/pyspark.ml.html#pyspark.ml.feature.RobustScaler) +for more details on the API. + +{% include_example python/ml/robust_scaler_example.py %} +
+
+ + ## MinMaxScaler `MinMaxScaler` transforms a dataset of `Vector` rows, rescaling each feature to a specific range (often [0, 1]). It takes parameters: diff --git a/examples/src/main/java/org/apache/spark/examples/ml/JavaRobustScalerExample.java b/examples/src/main/java/org/apache/spark/examples/ml/JavaRobustScalerExample.java new file mode 100644 index 0000000000000..475d046496d39 --- /dev/null +++ b/examples/src/main/java/org/apache/spark/examples/ml/JavaRobustScalerExample.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.examples.ml; + +import org.apache.spark.sql.SparkSession; + +// $example on$ +import org.apache.spark.ml.feature.RobustScaler; +import org.apache.spark.ml.feature.RobustScalerModel; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +// $example off$ + +public class JavaRobustScalerExample { + public static void main(String[] args) { + SparkSession spark = SparkSession + .builder() + .appName("JavaRobustScalerExample") + .getOrCreate(); + + // $example on$ + Dataset dataFrame = + spark.read().format("libsvm").load("data/mllib/sample_libsvm_data.txt"); + + RobustScaler scaler = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaledFeatures") + .setWithScaling(true) + .setWithCentering(false) + .setLower(0.25) + .setUpper(0.75); + + // Compute summary statistics by fitting the RobustScaler + RobustScalerModel scalerModel = scaler.fit(dataFrame); + + // Transform each feature to have unit quantile range. + Dataset scaledData = scalerModel.transform(dataFrame); + scaledData.show(); + // $example off$ + spark.stop(); + } +} diff --git a/examples/src/main/python/ml/robust_scaler_example.py b/examples/src/main/python/ml/robust_scaler_example.py new file mode 100644 index 0000000000000..435e9ccb806c6 --- /dev/null +++ b/examples/src/main/python/ml/robust_scaler_example.py @@ -0,0 +1,45 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from __future__ import print_function + +# $example on$ +from pyspark.ml.feature import RobustScaler +# $example off$ +from pyspark.sql import SparkSession + +if __name__ == "__main__": + spark = SparkSession\ + .builder\ + .appName("RobustScalerExample")\ + .getOrCreate() + + # $example on$ + dataFrame = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt") + scaler = RobustScaler(inputCol="features", outputCol="scaledFeatures", + withScaling=True, withCentering=False, + lower=0.25, upper=0.75) + + # Compute summary statistics by fitting the RobustScaler + scalerModel = scaler.fit(dataFrame) + + # Transform each feature to have unit quantile range. + scaledData = scalerModel.transform(dataFrame) + scaledData.show() + # $example off$ + + spark.stop() diff --git a/examples/src/main/scala/org/apache/spark/examples/ml/RobustScalerExample.scala b/examples/src/main/scala/org/apache/spark/examples/ml/RobustScalerExample.scala new file mode 100644 index 0000000000000..4f40c90dcaa38 --- /dev/null +++ b/examples/src/main/scala/org/apache/spark/examples/ml/RobustScalerExample.scala @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// scalastyle:off println +package org.apache.spark.examples.ml + +// $example on$ +import org.apache.spark.ml.feature.RobustScaler +// $example off$ +import org.apache.spark.sql.SparkSession + +object RobustScalerExample { + def main(args: Array[String]): Unit = { + val spark = SparkSession + .builder + .appName("RobustScalerExample") + .getOrCreate() + + // $example on$ + val dataFrame = spark.read.format("libsvm").load("data/mllib/sample_libsvm_data.txt") + + val scaler = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaledFeatures") + .setWithScaling(true) + .setWithCentering(false) + .setLower(0.25) + .setUpper(0.75) + + // Compute summary statistics by fitting the RobustScaler. + val scalerModel = scaler.fit(dataFrame) + + // Transform each feature to have unit quantile range. + val scaledData = scalerModel.transform(dataFrame) + scaledData.show() + // $example off$ + + spark.stop() + } +} +// scalastyle:on println diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/RobustScaler.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/RobustScaler.scala new file mode 100644 index 0000000000000..9dae39756d31e --- /dev/null +++ b/mllib/src/main/scala/org/apache/spark/ml/feature/RobustScaler.scala @@ -0,0 +1,288 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.ml.feature + +import org.apache.hadoop.fs.Path + +import org.apache.spark.annotation.Since +import org.apache.spark.ml.{Estimator, Model} +import org.apache.spark.ml.linalg._ +import org.apache.spark.ml.param._ +import org.apache.spark.ml.param.shared.{HasInputCol, HasOutputCol} +import org.apache.spark.ml.util._ +import org.apache.spark.mllib.util.MLUtils +import org.apache.spark.sql._ +import org.apache.spark.sql.catalyst.util.QuantileSummaries +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.types.{StructField, StructType} + +/** + * Params for [[RobustScaler]] and [[RobustScalerModel]]. + */ +private[feature] trait RobustScalerParams extends Params with HasInputCol with HasOutputCol { + + /** + * Lower quantile to calculate quantile range, shared by all features + * Default: 0.25 + * @group param + */ + val lower: DoubleParam = new DoubleParam(this, "lower", + "Lower quantile to calculate quantile range", + ParamValidators.inRange(0, 1, false, false)) + + /** @group getParam */ + def getLower: Double = $(lower) + + setDefault(lower -> 0.25) + + /** + * Upper quantile to calculate quantile range, shared by all features + * Default: 0.75 + * @group param + */ + val upper: DoubleParam = new DoubleParam(this, "upper", + "Upper quantile to calculate quantile range", + ParamValidators.inRange(0, 1, false, false)) + + /** @group getParam */ + def getUpper: Double = $(upper) + + setDefault(upper -> 0.75) + + /** + * Whether to center the data with median before scaling. + * It will build a dense output, so take care when applying to sparse input. + * Default: false + * @group param + */ + val withCentering: BooleanParam = new BooleanParam(this, "withCentering", + "Whether to center data with median") + + /** @group getParam */ + def getWithCentering: Boolean = $(withCentering) + + setDefault(withCentering -> false) + + /** + * Whether to scale the data to quantile range. + * Default: true + * @group param + */ + val withScaling: BooleanParam = new BooleanParam(this, "withScaling", + "Whether to scale the data to quantile range") + + /** @group getParam */ + def getWithScaling: Boolean = $(withScaling) + + setDefault(withScaling -> true) + + /** Validates and transforms the input schema. */ + protected def validateAndTransformSchema(schema: StructType): StructType = { + require($(lower) < $(upper), s"The specified lower quantile(${$(lower)}) is " + + s"larger or equal to upper quantile(${$(upper)})") + SchemaUtils.checkColumnType(schema, $(inputCol), new VectorUDT) + require(!schema.fieldNames.contains($(outputCol)), + s"Output column ${$(outputCol)} already exists.") + val outputFields = schema.fields :+ StructField($(outputCol), new VectorUDT, false) + StructType(outputFields) + } +} + + +/** + * Scale features using statistics that are robust to outliers. + * RobustScaler removes the median and scales the data according to the quantile range. + * The quantile range is by default IQR (Interquartile Range, quantile range between the + * 1st quartile = 25th quantile and the 3rd quartile = 75th quantile) but can be configured. + * Centering and scaling happen independently on each feature by computing the relevant + * statistics on the samples in the training set. Median and quantile range are then + * stored to be used on later data using the transform method. + * Standardization of a dataset is a common requirement for many machine learning estimators. + * Typically this is done by removing the mean and scaling to unit variance. However, + * outliers can often influence the sample mean / variance in a negative way. + * In such cases, the median and the quantile range often give better results. + */ +@Since("3.0.0") +class RobustScaler (override val uid: String) + extends Estimator[RobustScalerModel] with RobustScalerParams with DefaultParamsWritable { + + def this() = this(Identifiable.randomUID("robustScal")) + + /** @group setParam */ + def setInputCol(value: String): this.type = set(inputCol, value) + + /** @group setParam */ + def setOutputCol(value: String): this.type = set(outputCol, value) + + /** @group setParam */ + def setLower(value: Double): this.type = set(lower, value) + + /** @group setParam */ + def setUpper(value: Double): this.type = set(upper, value) + + /** @group setParam */ + def setWithCentering(value: Boolean): this.type = set(withCentering, value) + + /** @group setParam */ + def setWithScaling(value: Boolean): this.type = set(withScaling, value) + + override def fit(dataset: Dataset[_]): RobustScalerModel = { + transformSchema(dataset.schema, logging = true) + + val summaries = dataset.select($(inputCol)).rdd.map { + case Row(vec: Vector) => vec + }.mapPartitions { iter => + var agg: Array[QuantileSummaries] = null + while (iter.hasNext) { + val vec = iter.next() + if (agg == null) { + agg = Array.fill(vec.size)( + new QuantileSummaries(QuantileSummaries.defaultCompressThreshold, 0.001)) + } + require(vec.size == agg.length, + s"Number of dimensions must be ${agg.length} but got ${vec.size}") + var i = 0 + while (i < vec.size) { + agg(i) = agg(i).insert(vec(i)) + i += 1 + } + } + + if (agg == null) { + Iterator.empty + } else { + Iterator.single(agg.map(_.compress)) + } + }.treeReduce { (agg1, agg2) => + require(agg1.length == agg2.length) + var i = 0 + while (i < agg1.length) { + agg1(i) = agg1(i).merge(agg2(i)) + i += 1 + } + agg1 + } + + val (range, median) = summaries.map { s => + (s.query($(upper)).get - s.query($(lower)).get, + s.query(0.5).get) + }.unzip + + copyValues(new RobustScalerModel(uid, Vectors.dense(range).compressed, + Vectors.dense(median).compressed).setParent(this)) + } + + override def transformSchema(schema: StructType): StructType = { + validateAndTransformSchema(schema) + } + + override def copy(extra: ParamMap): RobustScaler = defaultCopy(extra) +} + +@Since("3.0.0") +object RobustScaler extends DefaultParamsReadable[RobustScaler] { + + override def load(path: String): RobustScaler = super.load(path) +} + +/** + * Model fitted by [[RobustScaler]]. + * + * @param range quantile range for each original column during fitting + * @param median median value for each original column during fitting + */ +@Since("3.0.0") +class RobustScalerModel private[ml] ( + override val uid: String, + val range: Vector, + val median: Vector) + extends Model[RobustScalerModel] with RobustScalerParams with MLWritable { + + import RobustScalerModel._ + + /** @group setParam */ + def setInputCol(value: String): this.type = set(inputCol, value) + + /** @group setParam */ + def setOutputCol(value: String): this.type = set(outputCol, value) + + override def transform(dataset: Dataset[_]): DataFrame = { + transformSchema(dataset.schema, logging = true) + + val shift = if ($(withCentering)) median.toArray else Array.emptyDoubleArray + val scale = if ($(withScaling)) { + range.toArray.map { v => if (v == 0) 0.0 else 1.0 / v } + } else Array.emptyDoubleArray + + val func = StandardScalerModel.getTransformFunc( + shift, scale, $(withCentering), $(withScaling)) + val transformer = udf(func) + + dataset.withColumn($(outputCol), transformer(col($(inputCol)))) + } + + override def transformSchema(schema: StructType): StructType = { + validateAndTransformSchema(schema) + } + + override def copy(extra: ParamMap): RobustScalerModel = { + val copied = new RobustScalerModel(uid, range, median) + copyValues(copied, extra).setParent(parent) + } + + override def write: MLWriter = new RobustScalerModelWriter(this) +} + +@Since("3.0.0") +object RobustScalerModel extends MLReadable[RobustScalerModel] { + + private[RobustScalerModel] + class RobustScalerModelWriter(instance: RobustScalerModel) extends MLWriter { + + private case class Data(range: Vector, median: Vector) + + override protected def saveImpl(path: String): Unit = { + DefaultParamsWriter.saveMetadata(instance, path, sc) + val data = new Data(instance.range, instance.median) + val dataPath = new Path(path, "data").toString + sparkSession.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath) + } + } + + private class RobustScalerModelReader extends MLReader[RobustScalerModel] { + + private val className = classOf[RobustScalerModel].getName + + override def load(path: String): RobustScalerModel = { + val metadata = DefaultParamsReader.loadMetadata(path, sc, className) + val dataPath = new Path(path, "data").toString + val data = sparkSession.read.parquet(dataPath) + val Row(range: Vector, median: Vector) = MLUtils + .convertVectorColumnsToML(data, "range", "median") + .select("range", "median") + .head() + val model = new RobustScalerModel(metadata.uid, range, median) + metadata.getAndSetParams(model) + model + } + } + + override def read: MLReader[RobustScalerModel] = new RobustScalerModelReader + + override def load(path: String): RobustScalerModel = super.load(path) +} diff --git a/mllib/src/main/scala/org/apache/spark/ml/feature/StandardScaler.scala b/mllib/src/main/scala/org/apache/spark/ml/feature/StandardScaler.scala index 81cf2e1a4ff79..01be781ec5aad 100644 --- a/mllib/src/main/scala/org/apache/spark/ml/feature/StandardScaler.scala +++ b/mllib/src/main/scala/org/apache/spark/ml/feature/StandardScaler.scala @@ -25,11 +25,10 @@ import org.apache.spark.ml.linalg._ import org.apache.spark.ml.param._ import org.apache.spark.ml.param.shared._ import org.apache.spark.ml.util._ -import org.apache.spark.mllib.feature -import org.apache.spark.mllib.linalg.{Vector => OldVector, Vectors => OldVectors} +import org.apache.spark.mllib.feature.{StandardScaler => OldStandardScaler} +import org.apache.spark.mllib.linalg.{Vectors => OldVectors} import org.apache.spark.mllib.linalg.VectorImplicits._ import org.apache.spark.mllib.util.MLUtils -import org.apache.spark.rdd.RDD import org.apache.spark.sql._ import org.apache.spark.sql.functions._ import org.apache.spark.sql.types.{StructField, StructType} @@ -110,12 +109,13 @@ class StandardScaler @Since("1.4.0") ( @Since("2.0.0") override def fit(dataset: Dataset[_]): StandardScalerModel = { transformSchema(dataset.schema, logging = true) - val input: RDD[OldVector] = dataset.select($(inputCol)).rdd.map { + val input = dataset.select($(inputCol)).rdd.map { case Row(v: Vector) => OldVectors.fromML(v) } - val scaler = new feature.StandardScaler(withMean = $(withMean), withStd = $(withStd)) + val scaler = new OldStandardScaler(withMean = $(withMean), withStd = $(withStd)) val scalerModel = scaler.fit(input) - copyValues(new StandardScalerModel(uid, scalerModel.std, scalerModel.mean).setParent(this)) + copyValues(new StandardScalerModel(uid, scalerModel.std.compressed, + scalerModel.mean.compressed).setParent(this)) } @Since("1.4.0") @@ -160,35 +160,14 @@ class StandardScalerModel private[ml] ( @Since("2.0.0") override def transform(dataset: Dataset[_]): DataFrame = { transformSchema(dataset.schema, logging = true) - val scaler = new feature.StandardScalerModel(std, mean, $(withStd), $(withMean)) - - val func = if ($(withMean)) { - vector: Vector => - val values = vector match { - // specially handle DenseVector because its toArray does not clone already - case d: DenseVector => d.values.clone() - case v: Vector => v.toArray - } - val newValues = scaler.transformWithMean(values) - Vectors.dense(newValues) - } else if ($(withStd)) { - vector: Vector => - vector match { - case DenseVector(values) => - val newValues = scaler.transformDenseWithStd(values) - Vectors.dense(newValues) - case SparseVector(size, indices, values) => - val (newIndices, newValues) = scaler.transformSparseWithStd(indices, values) - Vectors.sparse(size, newIndices, newValues) - case other => - throw new UnsupportedOperationException( - s"Only sparse and dense vectors are supported but got ${other.getClass}.") - } - } else { - vector: Vector => vector - } + val shift = if ($(withMean)) mean.toArray else Array.emptyDoubleArray + val scale = if ($(withStd)) { + std.toArray.map { v => if (v == 0) 0.0 else 1.0 / v } + } else Array.emptyDoubleArray + val func = getTransformFunc(shift, scale, $(withMean), $(withStd)) val transformer = udf(func) + dataset.withColumn($(outputCol), transformer(col($(inputCol)))) } @@ -245,4 +224,90 @@ object StandardScalerModel extends MLReadable[StandardScalerModel] { @Since("1.6.0") override def load(path: String): StandardScalerModel = super.load(path) + + private[spark] def transformWithBoth( + shift: Array[Double], + scale: Array[Double], + values: Array[Double]): Array[Double] = { + var i = 0 + while (i < values.length) { + values(i) = (values(i) - shift(i)) * scale(i) + i += 1 + } + values + } + + private[spark] def transformWithShift( + shift: Array[Double], + values: Array[Double]): Array[Double] = { + var i = 0 + while (i < values.length) { + values(i) -= shift(i) + i += 1 + } + values + } + + private[spark] def transformDenseWithScale( + scale: Array[Double], + values: Array[Double]): Array[Double] = { + var i = 0 + while (i < values.length) { + values(i) *= scale(i) + i += 1 + } + values + } + + private[spark] def transformSparseWithScale( + scale: Array[Double], + indices: Array[Int], + values: Array[Double]): Array[Double] = { + var i = 0 + while (i < values.length) { + values(i) *= scale(indices(i)) + i += 1 + } + values + } + + private[ml] def getTransformFunc( + shift: Array[Double], + scale: Array[Double], + withShift: Boolean, + withScale: Boolean): Vector => Vector = { + (withShift, withScale) match { + case (true, true) => + vector: Vector => + val values = vector match { + case d: DenseVector => d.values.clone() + case v: Vector => v.toArray + } + val newValues = transformWithBoth(shift, scale, values) + Vectors.dense(newValues) + + case (true, false) => + vector: Vector => + val values = vector match { + case d: DenseVector => d.values.clone() + case v: Vector => v.toArray + } + val newValues = transformWithShift(shift, values) + Vectors.dense(newValues) + + case (false, true) => + vector: Vector => + vector match { + case DenseVector(values) => + val newValues = transformDenseWithScale(scale, values.clone()) + Vectors.dense(newValues) + case SparseVector(size, indices, values) => + val newValues = transformSparseWithScale(scale, indices, values.clone()) + Vectors.sparse(size, indices, newValues) + } + + case (false, false) => + vector: Vector => vector + } + } } diff --git a/mllib/src/main/scala/org/apache/spark/mllib/feature/StandardScaler.scala b/mllib/src/main/scala/org/apache/spark/mllib/feature/StandardScaler.scala index 19e53e7eac844..7286733934ad9 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/feature/StandardScaler.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/feature/StandardScaler.scala @@ -19,6 +19,7 @@ package org.apache.spark.mllib.feature import org.apache.spark.annotation.{DeveloperApi, Since} import org.apache.spark.internal.Logging +import org.apache.spark.ml.feature.{StandardScalerModel => NewStandardScalerModel} import org.apache.spark.mllib.linalg.{DenseVector, SparseVector, Vector, Vectors} import org.apache.spark.mllib.stat.MultivariateOnlineSummarizer import org.apache.spark.rdd.RDD @@ -122,7 +123,8 @@ class StandardScalerModel @Since("1.3.0") ( // Since `shift` will be only used in `withMean` branch, we have it as // `lazy val` so it will be evaluated in that branch. Note that we don't // want to create this array multiple times in `transform` function. - private lazy val shift: Array[Double] = mean.toArray + private lazy val shift = mean.toArray + private lazy val scale = std.toArray.map { v => if (v == 0) 0.0 else 1.0 / v } /** * Applies standardization transformation on a vector. @@ -134,77 +136,49 @@ class StandardScalerModel @Since("1.3.0") ( @Since("1.1.0") override def transform(vector: Vector): Vector = { require(mean.size == vector.size) - if (withMean) { - // Must have a copy of the values since it will be modified in place - val values = vector match { - // specially handle DenseVector because its toArray does not clone already - case d: DenseVector => d.values.clone() - case v: Vector => v.toArray - } - val newValues = transformWithMean(values) - Vectors.dense(newValues) - } else if (withStd) { - vector match { - case DenseVector(values) => - val newValues = transformDenseWithStd(values) - Vectors.dense(newValues) - case SparseVector(size, indices, values) => - val (newIndices, newValues) = transformSparseWithStd(indices, values) - Vectors.sparse(size, newIndices, newValues) - case other => - throw new UnsupportedOperationException( - s"Only sparse and dense vectors are supported but got ${other.getClass}.") - } - } else { - // Note that it's safe since we always assume that the data in RDD should be immutable. - vector - } - } - - private[spark] def transformWithMean(values: Array[Double]): Array[Double] = { - // By default, Scala generates Java methods for member variables. So every time when - // the member variables are accessed, `invokespecial` will be called which is expensive. - // This can be avoid by having a local reference of `shift`. - val localShift = shift - val size = values.length - if (withStd) { - var i = 0 - while (i < size) { - values(i) = if (std(i) != 0.0) (values(i) - localShift(i)) * (1.0 / std(i)) else 0.0 - i += 1 - } - } else { - var i = 0 - while (i < size) { - values(i) -= localShift(i) - i += 1 - } - } - values - } - - private[spark] def transformDenseWithStd(values: Array[Double]): Array[Double] = { - val size = values.length - val newValues = values.clone() - var i = 0 - while(i < size) { - newValues(i) *= (if (std(i) != 0.0) 1.0 / std(i) else 0.0) - i += 1 - } - newValues - } - private[spark] def transformSparseWithStd(indices: Array[Int], - values: Array[Double]): (Array[Int], Array[Double]) = { - // For sparse vector, the `index` array inside sparse vector object will not be changed, - // so we can re-use it to save memory. - val nnz = values.length - val newValues = values.clone() - var i = 0 - while (i < nnz) { - newValues(i) *= (if (std(indices(i)) != 0.0) 1.0 / std(indices(i)) else 0.0) - i += 1 + (withMean, withStd) match { + case (true, true) => + // By default, Scala generates Java methods for member variables. So every time when + // the member variables are accessed, `invokespecial` will be called which is expensive. + // This can be avoid by having a local reference of `shift`. + val localShift = shift + val localScale = scale + val values = vector match { + // specially handle DenseVector because its toArray does not clone already + case d: DenseVector => d.values.clone() + case v: Vector => v.toArray + } + val newValues = NewStandardScalerModel + .transformWithBoth(localShift, localScale, values) + Vectors.dense(newValues) + + case (true, false) => + val localShift = shift + val values = vector match { + case d: DenseVector => d.values.clone() + case v: Vector => v.toArray + } + val newValues = NewStandardScalerModel + .transformWithShift(localShift, values) + Vectors.dense(newValues) + + case (false, true) => + val localScale = scale + vector match { + case DenseVector(values) => + val newValues = NewStandardScalerModel + .transformDenseWithScale(localScale, values.clone()) + Vectors.dense(newValues) + case SparseVector(size, indices, values) => + // For sparse vector, the `index` array inside sparse vector object will not be changed, + // so we can re-use it to save memory. + val newValues = NewStandardScalerModel + .transformSparseWithScale(localScale, indices, values.clone()) + Vectors.sparse(size, indices, newValues) + } + + case _ => vector } - (indices, newValues) } } diff --git a/mllib/src/test/scala/org/apache/spark/ml/feature/RobustScalerSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/feature/RobustScalerSuite.scala new file mode 100644 index 0000000000000..335f144e748e4 --- /dev/null +++ b/mllib/src/test/scala/org/apache/spark/ml/feature/RobustScalerSuite.scala @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.ml.feature + +import org.apache.spark.ml.linalg.{Vector, Vectors} +import org.apache.spark.ml.param.ParamsSuite +import org.apache.spark.ml.util.{DefaultReadWriteTest, MLTest, MLTestingUtils} +import org.apache.spark.ml.util.TestingUtils._ +import org.apache.spark.sql.Row + +class RobustScalerSuite extends MLTest with DefaultReadWriteTest { + + import testImplicits._ + + @transient var data: Array[Vector] = _ + @transient var resWithScaling: Array[Vector] = _ + @transient var resWithCentering: Array[Vector] = _ + @transient var resWithBoth: Array[Vector] = _ + + override def beforeAll(): Unit = { + super.beforeAll() + + // median = [2.0, -2.0] + // 1st quartile = [1.0, -3.0] + // 3st quartile = [3.0, -1.0] + // quantile range = IQR = [2.0, 2.0] + data = Array( + Vectors.dense(0.0, 0.0), + Vectors.dense(1.0, -1.0), + Vectors.dense(2.0, -2.0), + Vectors.dense(3.0, -3.0), + Vectors.dense(4.0, -4.0) + ) + + /* + Using the following Python code to load the data and train the model using + scikit-learn package. + + from sklearn.preprocessing import RobustScaler + import numpy as np + X = np.array([[0, 0], [1, -1], [2, -2], [3, -3], [4, -4]], dtype=np.float) + scaler = RobustScaler(with_centering=True, with_scaling=False).fit(X) + + >>> scaler.center_ + array([ 2., -2.]) + >>> scaler.scale_ + array([2., 2.]) + >>> scaler.transform(X) + array([[-2., 2.], + [-1., 1.], + [ 0., 0.], + [ 1., -1.], + [ 2., -2.]]) + */ + resWithCentering = Array( + Vectors.dense(-2.0, 2.0), + Vectors.dense(-1.0, 1.0), + Vectors.dense(0.0, 0.0), + Vectors.dense(1.0, -1.0), + Vectors.dense(2.0, -2.0) + ) + + /* + Python code: + + scaler = RobustScaler(with_centering=False, with_scaling=True).fit(X) + >>> scaler.transform(X) + array([[ 0. , 0. ], + [ 0.5, -0.5], + [ 1. , -1. ], + [ 1.5, -1.5], + [ 2. , -2. ]]) + */ + resWithScaling = Array( + Vectors.dense(0.0, 0.0), + Vectors.dense(0.5, -0.5), + Vectors.dense(1.0, -1.0), + Vectors.dense(1.5, -1.5), + Vectors.dense(2.0, -2.0) + ) + + /* + Python code: + + scaler = RobustScaler(with_centering=True, with_scaling=True).fit(X) + >>> scaler.transform(X) + array([[-1. , 1. ], + [-0.5, 0.5], + [ 0. , 0. ], + [ 0.5, -0.5], + [ 1. , -1. ]]) + */ + resWithBoth = Array( + Vectors.dense(-1.0, 1.0), + Vectors.dense(-0.5, 0.5), + Vectors.dense(0.0, 0.0), + Vectors.dense(0.5, -0.5), + Vectors.dense(1.0, -1.0) + ) + } + + + private def assertResult: Row => Unit = { + case Row(vector1: Vector, vector2: Vector) => + assert(vector1 ~== vector2 absTol 1E-5, + "The vector value is not correct after transformation.") + } + + test("params") { + ParamsSuite.checkParams(new RobustScaler) + ParamsSuite.checkParams(new RobustScalerModel("empty", + Vectors.dense(1.0), Vectors.dense(2.0))) + } + + test("Scaling with default parameter") { + val df0 = data.zip(resWithScaling).toSeq.toDF("features", "expected") + + val robustScalerEst0 = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaled_features") + val robustScaler0 = robustScalerEst0.fit(df0) + MLTestingUtils.checkCopyAndUids(robustScalerEst0, robustScaler0) + + testTransformer[(Vector, Vector)](df0, robustScaler0, "scaled_features", "expected")( + assertResult) + } + + test("Scaling with setter") { + val df1 = data.zip(resWithBoth).toSeq.toDF("features", "expected") + val df2 = data.zip(resWithCentering).toSeq.toDF("features", "expected") + val df3 = data.zip(data).toSeq.toDF("features", "expected") + + val robustScaler1 = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaled_features") + .setWithCentering(true) + .setWithScaling(true) + .fit(df1) + + val robustScaler2 = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaled_features") + .setWithCentering(true) + .setWithScaling(false) + .fit(df2) + + val robustScaler3 = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaled_features") + .setWithCentering(false) + .setWithScaling(false) + .fit(df3) + + testTransformer[(Vector, Vector)](df1, robustScaler1, "scaled_features", "expected")( + assertResult) + testTransformer[(Vector, Vector)](df2, robustScaler2, "scaled_features", "expected")( + assertResult) + testTransformer[(Vector, Vector)](df3, robustScaler3, "scaled_features", "expected")( + assertResult) + } + + test("sparse data and withCentering") { + val someSparseData = data.zipWithIndex.map { + case (vec, i) => if (i % 2 == 0) vec.toSparse else vec + } + val df = someSparseData.zip(resWithCentering).toSeq.toDF("features", "expected") + val robustScaler = new RobustScaler() + .setInputCol("features") + .setOutputCol("scaled_features") + .setWithCentering(true) + .setWithScaling(false) + .fit(df) + testTransformer[(Vector, Vector)](df, robustScaler, "scaled_features", "expected")( + assertResult) + } + + test("RobustScaler read/write") { + val t = new RobustScaler() + .setInputCol("myInputCol") + .setOutputCol("myOutputCol") + .setWithCentering(false) + .setWithScaling(true) + testDefaultReadWrite(t) + } + + test("RobustScalerModel read/write") { + val instance = new RobustScalerModel("myRobustScalerModel", + Vectors.dense(1.0, 2.0), Vectors.dense(3.0, 4.0)) + val newInstance = testDefaultReadWrite(instance) + assert(newInstance.range === instance.range) + assert(newInstance.median === instance.median) + } + +} diff --git a/mllib/src/test/scala/org/apache/spark/ml/feature/StandardScalerSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/feature/StandardScalerSuite.scala index c5c49d67194e4..07645b36153c7 100644 --- a/mllib/src/test/scala/org/apache/spark/ml/feature/StandardScalerSuite.scala +++ b/mllib/src/test/scala/org/apache/spark/ml/feature/StandardScalerSuite.scala @@ -21,7 +21,7 @@ import org.apache.spark.ml.linalg.{Vector, Vectors} import org.apache.spark.ml.param.ParamsSuite import org.apache.spark.ml.util.{DefaultReadWriteTest, MLTest, MLTestingUtils} import org.apache.spark.ml.util.TestingUtils._ -import org.apache.spark.sql.{DataFrame, Row} +import org.apache.spark.sql.Row class StandardScalerSuite extends MLTest with DefaultReadWriteTest { @@ -57,7 +57,7 @@ class StandardScalerSuite extends MLTest with DefaultReadWriteTest { ) } - def assertResult: Row => Unit = { + private def assertResult: Row => Unit = { case Row(vector1: Vector, vector2: Vector) => assert(vector1 ~== vector2 absTol 1E-5, "The vector value is not correct after standardization.") diff --git a/python/pyspark/ml/feature.py b/python/pyspark/ml/feature.py index 78d02690c4d46..2df080c30b046 100755 --- a/python/pyspark/ml/feature.py +++ b/python/pyspark/ml/feature.py @@ -49,6 +49,7 @@ 'PCA', 'PCAModel', 'PolynomialExpansion', 'QuantileDiscretizer', + 'RobustScaler', 'RobustScalerModel', 'RegexTokenizer', 'RFormula', 'RFormulaModel', 'SQLTransformer', @@ -2037,6 +2038,167 @@ def _create_model(self, java_model): handleInvalid=self.getHandleInvalid()) +@inherit_doc +class RobustScaler(JavaEstimator, HasInputCol, HasOutputCol, JavaMLReadable, JavaMLWritable): + """ + RobustScaler removes the median and scales the data according to the quantile range. + The quantile range is by default IQR (Interquartile Range, quantile range between the + 1st quartile = 25th quantile and the 3rd quartile = 75th quantile) but can be configured. + Centering and scaling happen independently on each feature by computing the relevant + statistics on the samples in the training set. Median and quantile range are then + stored to be used on later data using the transform method. + + >>> from pyspark.ml.linalg import Vectors + >>> data = [(0, Vectors.dense([0.0, 0.0]),), + ... (1, Vectors.dense([1.0, -1.0]),), + ... (2, Vectors.dense([2.0, -2.0]),), + ... (3, Vectors.dense([3.0, -3.0]),), + ... (4, Vectors.dense([4.0, -4.0]),),] + >>> df = spark.createDataFrame(data, ["id", "features"]) + >>> scaler = RobustScaler(inputCol="features", outputCol="scaled") + >>> model = scaler.fit(df) + >>> model.median + DenseVector([2.0, -2.0]) + >>> model.range + DenseVector([2.0, 2.0]) + >>> model.transform(df).collect()[1].scaled + DenseVector([0.5, -0.5]) + >>> scalerPath = temp_path + "/robust-scaler" + >>> scaler.save(scalerPath) + >>> loadedScaler = RobustScaler.load(scalerPath) + >>> loadedScaler.getWithCentering() == scaler.getWithCentering() + True + >>> loadedScaler.getWithScaling() == scaler.getWithScaling() + True + >>> modelPath = temp_path + "/robust-scaler-model" + >>> model.save(modelPath) + >>> loadedModel = RobustScalerModel.load(modelPath) + >>> loadedModel.median == model.median + True + >>> loadedModel.range == model.range + True + + .. versionadded:: 3.0.0 + """ + + lower = Param(Params._dummy(), "lower", "Lower quantile to calculate quantile range", + typeConverter=TypeConverters.toFloat) + upper = Param(Params._dummy(), "upper", "Upper quantile to calculate quantile range", + typeConverter=TypeConverters.toFloat) + withCentering = Param(Params._dummy(), "withCentering", "Whether to center data with median", + typeConverter=TypeConverters.toBoolean) + withScaling = Param(Params._dummy(), "withScaling", "Whether to scale the data to " + "quantile range", typeConverter=TypeConverters.toBoolean) + + @keyword_only + def __init__(self, lower=0.25, upper=0.75, withCentering=False, withScaling=True, + inputCol=None, outputCol=None): + """ + __init__(self, lower=0.25, upper=0.75, withCentering=False, withScaling=True, \ + inputCol=None, outputCol=None) + """ + super(RobustScaler, self).__init__() + self._java_obj = self._new_java_obj("org.apache.spark.ml.feature.RobustScaler", self.uid) + self._setDefault(lower=0.25, upper=0.75, withCentering=False, withScaling=True) + kwargs = self._input_kwargs + self.setParams(**kwargs) + + @keyword_only + @since("3.0.0") + def setParams(self, lower=0.25, upper=0.75, withCentering=False, withScaling=True, + inputCol=None, outputCol=None): + """ + setParams(self, lower=0.25, upper=0.75, withCentering=False, withScaling=True, \ + inputCol=None, outputCol=None) + Sets params for this RobustScaler. + """ + kwargs = self._input_kwargs + return self._set(**kwargs) + + @since("3.0.0") + def setLower(self, value): + """ + Sets the value of :py:attr:`lower`. + """ + return self._set(lower=value) + + @since("3.0.0") + def getLower(self): + """ + Gets the value of lower or its default value. + """ + return self.getOrDefault(self.lower) + + @since("3.0.0") + def setUpper(self, value): + """ + Sets the value of :py:attr:`upper`. + """ + return self._set(upper=value) + + @since("3.0.0") + def getUpper(self): + """ + Gets the value of upper or its default value. + """ + return self.getOrDefault(self.upper) + + @since("3.0.0") + def setWithCentering(self, value): + """ + Sets the value of :py:attr:`withCentering`. + """ + return self._set(withCentering=value) + + @since("3.0.0") + def getWithCentering(self): + """ + Gets the value of withCentering or its default value. + """ + return self.getOrDefault(self.withCentering) + + @since("3.0.0") + def setWithScaling(self, value): + """ + Sets the value of :py:attr:`withScaling`. + """ + return self._set(withScaling=value) + + @since("3.0.0") + def getWithScaling(self): + """ + Gets the value of withScaling or its default value. + """ + return self.getOrDefault(self.withScaling) + + def _create_model(self, java_model): + return RobustScalerModel(java_model) + + +class RobustScalerModel(JavaModel, JavaMLReadable, JavaMLWritable): + """ + Model fitted by :py:class:`RobustScaler`. + + .. versionadded:: 3.0.0 + """ + + @property + @since("3.0.0") + def median(self): + """ + Median of the RobustScalerModel. + """ + return self._call_java("median") + + @property + @since("3.0.0") + def range(self): + """ + Quantile range of the RobustScalerModel. + """ + return self._call_java("range") + + @inherit_doc @ignore_unicode_prefix class RegexTokenizer(JavaTransformer, HasInputCol, HasOutputCol, JavaMLReadable, JavaMLWritable):