[SPARK-5207] [MLLIB] [WIP] change StandardScalerModel contructor to not be private to mllib, added tests for newly-exposed functionality

Author: ogeagla

mllib/src/main/scala/org/apache/spark/mllib/feature/StandardScaler.scala

@@ -53,28 +53,38 @@ class StandardScaler(withMean: Boolean, withStd: Boolean) extends Logging {
     val summary = data.treeAggregate(new MultivariateOnlineSummarizer)(
       (aggregator, data) => aggregator.add(data),
       (aggregator1, aggregator2) => aggregator1.merge(aggregator2))
-    new StandardScalerModel(withMean, withStd, summary.mean, summary.variance)
+    new StandardScalerModel(summary.mean, summary.variance, withMean, withStd)
   }
 }
 
 /**
  * :: Experimental ::
  * Represents a StandardScaler model that can transform vectors.
  *
- * @param withMean whether to center the data before scaling
- * @param withStd whether to scale the data to have unit standard deviation
  * @param mean column mean values
  * @param variance column variance values
+ * @param withMean whether to center the data before scaling
+ * @param withStd whether to scale the data to have unit standard deviation
  */
 @Experimental
-class StandardScalerModel private[mllib] (
-    val withMean: Boolean,
-    val withStd: Boolean,
+class StandardScalerModel (
     val mean: Vector,
-    val variance: Vector) extends VectorTransformer {
+    val variance: Vector,
+    private var withMean: Boolean = false,
+    private var withStd: Boolean = true) extends VectorTransformer {
 
   require(mean.size == variance.size)
 
+  def setWithMean(withMean: Boolean): this.type = {
+    this.withMean = withMean
+    this
+  }
+
+  def setWithStd(withStd: Boolean): this.type = {
+    this.withStd = withStd
+    this
+  }
+
   private lazy val factor: Array[Double] = {
     val f = Array.ofDim[Double](variance.size)
     var i = 0

mllib/src/test/scala/org/apache/spark/mllib/feature/StandardScalerSuite.scala

@@ -28,58 +28,88 @@ import org.apache.spark.rdd.RDD
 
 class StandardScalerSuite extends FunSuite with MLlibTestSparkContext {
 
-  private def computeSummary(data: RDD[Vector]): MultivariateStatisticalSummary = {
-    data.treeAggregate(new MultivariateOnlineSummarizer)(
-      (aggregator, data) => aggregator.add(data),
-      (aggregator1, aggregator2) => aggregator1.merge(aggregator2))
+  // When the input data is all constant, the variance is zero. The standardization against
+  // zero variance is not well-defined, but we decide to just set it into zero here.
+  val constantData = Array(
+    Vectors.dense(2.0),
+    Vectors.dense(2.0),
+    Vectors.dense(2.0)
+  )
+
+  val sparseData = Array(
+    Vectors.sparse(3, Seq((0, -2.0), (1, 2.3))),
+    Vectors.sparse(3, Seq((1, -1.0), (2, -3.0))),
+    Vectors.sparse(3, Seq((1, -5.1))),
+    Vectors.sparse(3, Seq((0, 3.8), (2, 1.9))),
+    Vectors.sparse(3, Seq((0, 1.7), (1, -0.6))),
+    Vectors.sparse(3, Seq((1, 1.9)))
+  )
+
+  val denseData = Array(
+    Vectors.dense(-2.0, 2.3, 0),
+    Vectors.dense(0.0, -1.0, -3.0),
+    Vectors.dense(0.0, -5.1, 0.0),
+    Vectors.dense(3.8, 0.0, 1.9),
+    Vectors.dense(1.7, -0.6, 0.0),
+    Vectors.dense(0.0, 1.9, 0.0)
+  )
+
+  def validateConstant(data1: Array[Vector], data2: Array[Vector], data3: Array[Vector]) {
+    assert(data1.forall(_.toArray.forall(_ == 0.0)),
+      "The variance is zero, so the transformed result should be 0.0")
+    assert(data2.forall(_.toArray.forall(_ == 0.0)),
+      "The variance is zero, so the transformed result should be 0.0")
+    assert(data3.forall(_.toArray.forall(_ == 0.0)),
+      "The variance is zero, so the transformed result should be 0.0")
   }
 
-  test("Standardization with dense input") {
-    val data = Array(
-      Vectors.dense(-2.0, 2.3, 0),
-      Vectors.dense(0.0, -1.0, -3.0),
-      Vectors.dense(0.0, -5.1, 0.0),
-      Vectors.dense(3.8, 0.0, 1.9),
-      Vectors.dense(1.7, -0.6, 0.0),
-      Vectors.dense(0.0, 1.9, 0.0)
-    )
+  def validateSparse(data: Array[Vector], dataRDD: RDD[Vector]) = {
 
-    val dataRDD = sc.parallelize(data, 3)
+    val summary = computeSummary(dataRDD)
 
-    val standardizer1 = new StandardScaler(withMean = true, withStd = true)
-    val standardizer2 = new StandardScaler()
-    val standardizer3 = new StandardScaler(withMean = true, withStd = false)
+    assert((sparseData, data, dataRDD.collect()).zipped.forall {
+      case (v1: DenseVector, v2: DenseVector, v3: DenseVector) => true
+      case (v1: SparseVector, v2: SparseVector, v3: SparseVector) => true
+      case _ => false
+    }, "The vector type should be preserved after standardization.")
 
-    val model1 = standardizer1.fit(dataRDD)
-    val model2 = standardizer2.fit(dataRDD)
-    val model3 = standardizer3.fit(dataRDD)
+    assert((data, dataRDD.collect()).zipped.forall((v1, v2) => v1 ~== v2 absTol 1E-5))
 
-    val data1 = data.map(model1.transform)
-    val data2 = data.map(model2.transform)
-    val data3 = data.map(model3.transform)
+    assert(summary.mean !~== Vectors.dense(0.0, 0.0, 0.0) absTol 1E-5)
+    assert(summary.variance ~== Vectors.dense(1.0, 1.0, 1.0) absTol 1E-5)
 
-    val data1RDD = model1.transform(dataRDD)
-    val data2RDD = model2.transform(dataRDD)
-    val data3RDD = model3.transform(dataRDD)
+    assert(data(4) ~== Vectors.sparse(3, Seq((0, 0.865538862), (1, -0.22604255))) absTol 1E-5)
+    assert(data(5) ~== Vectors.sparse(3, Seq((1, 0.71580142))) absTol 1E-5)
+  }
 
-    val summary = computeSummary(dataRDD)
+  def validateDense(
+      data1: Array[Vector],
+      data2: Array[Vector],
+      data3: Array[Vector],
+      refDataRDD: RDD[Vector],
+      data1RDD: RDD[Vector],
+      data2RDD: RDD[Vector],
+      data3RDD: RDD[Vector]
+      ) = {
+
+    val refSummary = computeSummary(refDataRDD)
     val summary1 = computeSummary(data1RDD)
     val summary2 = computeSummary(data2RDD)
     val summary3 = computeSummary(data3RDD)
 
-    assert((data, data1, data1RDD.collect()).zipped.forall {
+    assert((denseData, data1, data1RDD.collect()).zipped.forall {
       case (v1: DenseVector, v2: DenseVector, v3: DenseVector) => true
       case (v1: SparseVector, v2: SparseVector, v3: SparseVector) => true
       case _ => false
     }, "The vector type should be preserved after standardization.")
 
-    assert((data, data2, data2RDD.collect()).zipped.forall {
+    assert((denseData, data2, data2RDD.collect()).zipped.forall {
       case (v1: DenseVector, v2: DenseVector, v3: DenseVector) => true
       case (v1: SparseVector, v2: SparseVector, v3: SparseVector) => true
       case _ => false
     }, "The vector type should be preserved after standardization.")
 
-    assert((data, data3, data3RDD.collect()).zipped.forall {
+    assert((denseData, data3, data3RDD.collect()).zipped.forall {
       case (v1: DenseVector, v2: DenseVector, v3: DenseVector) => true
       case (v1: SparseVector, v2: SparseVector, v3: SparseVector) => true
       case _ => false
@@ -96,7 +126,7 @@ class StandardScalerSuite extends FunSuite with MLlibTestSparkContext {
     assert(summary2.variance ~== Vectors.dense(1.0, 1.0, 1.0) absTol 1E-5)
 
     assert(summary3.mean ~== Vectors.dense(0.0, 0.0, 0.0) absTol 1E-5)
-    assert(summary3.variance ~== summary.variance absTol 1E-5)
+    assert(summary3.variance ~== refSummary.variance absTol 1E-5)
 
     assert(data1(0) ~== Vectors.dense(-1.31527964, 1.023470449, 0.11637768424) absTol 1E-5)
     assert(data1(3) ~== Vectors.dense(1.637735298, 0.156973995, 1.32247368462) absTol 1E-5)
@@ -106,18 +136,103 @@ class StandardScalerSuite extends FunSuite with MLlibTestSparkContext {
     assert(data3(5) ~== Vectors.dense(-0.58333333, 2.316666666, 0.18333333333) absTol 1E-5)
   }
 
+  private def computeSummary(data: RDD[Vector]): MultivariateStatisticalSummary = {
+    data.treeAggregate(new MultivariateOnlineSummarizer)(
+      (aggregator, data) => aggregator.add(data),
+      (aggregator1, aggregator2) => aggregator1.merge(aggregator2))
+  }
+
+  test("Standardization with dense input when means and variances are provided") {
+
+    val dataRDD = sc.parallelize(denseData, 3)
+
+    val standardizer1 = new StandardScaler(withMean = true, withStd = true)
+    val standardizer2 = new StandardScaler()
+    val standardizer3 = new StandardScaler(withMean = true, withStd = false)
+
+    val model1 = standardizer1.fit(dataRDD)
+    val model2 = standardizer2.fit(dataRDD)
+    val model3 = standardizer3.fit(dataRDD)
+
+    val equivalentModel1 = new StandardScalerModel(model1.mean, model1.variance, true, true)
+    val equivalentModel2 = new StandardScalerModel(model2.mean, model2.variance)
+    val equivalentModel3 = new StandardScalerModel(model3.mean, model3.variance, true, false)
+
+    val data1 = denseData.map(equivalentModel1.transform)
+    val data2 = denseData.map(equivalentModel2.transform)
+    val data3 = denseData.map(equivalentModel3.transform)
+
+    val data1RDD = equivalentModel1.transform(dataRDD)
+    val data2RDD = equivalentModel2.transform(dataRDD)
+    val data3RDD = equivalentModel3.transform(dataRDD)
+
+    validateDense(data1, data2, data3, dataRDD, data1RDD, data2RDD, data3RDD)
+  }
+
+  test("Standardization with dense input") {
+
+    val dataRDD = sc.parallelize(denseData, 3)
+
+    val standardizer1 = new StandardScaler(withMean = true, withStd = true)
+    val standardizer2 = new StandardScaler()
+    val standardizer3 = new StandardScaler(withMean = true, withStd = false)
+
+    val model1 = standardizer1.fit(dataRDD)
+    val model2 = standardizer2.fit(dataRDD)
+    val model3 = standardizer3.fit(dataRDD)
+
+    val data1 = denseData.map(model1.transform)
+    val data2 = denseData.map(model2.transform)
+    val data3 = denseData.map(model3.transform)
+
+    val data1RDD = model1.transform(dataRDD)
+    val data2RDD = model2.transform(dataRDD)
+    val data3RDD = model3.transform(dataRDD)
+
+    validateDense(data1, data2, data3, dataRDD, data1RDD, data2RDD, data3RDD)
+  }
+
+
+  test("Standardization with sparse input when means and variances are provided") {
+
+    val dataRDD = sc.parallelize(sparseData, 3)
+
+    val standardizer1 = new StandardScaler(withMean = true, withStd = true)
+    val standardizer2 = new StandardScaler()
+    val standardizer3 = new StandardScaler(withMean = true, withStd = false)
+
+    val model1 = standardizer1.fit(dataRDD)
+    val model2 = standardizer2.fit(dataRDD)
+    val model3 = standardizer3.fit(dataRDD)
+
+    val equivalentModel1 = new StandardScalerModel(model1.mean, model1.variance, true, true)
+    val equivalentModel2 = new StandardScalerModel(model2.mean, model2.variance)
+    val equivalentModel3 = new StandardScalerModel(model3.mean, model3.variance, true, false)
+
+
+    val data2 = sparseData.map(equivalentModel2.transform)
+
+    withClue("Standardization with mean can not be applied on sparse input.") {
+      intercept[IllegalArgumentException] {
+        sparseData.map(equivalentModel1.transform)
+      }
+    }
+
+    withClue("Standardization with mean can not be applied on sparse input.") {
+      intercept[IllegalArgumentException] {
+        sparseData.map(equivalentModel3.transform)
+      }
+    }
+
+    val data2RDD = equivalentModel2.transform(dataRDD)
+
+    validateSparse(data2, data2RDD)
+
+  }
 
   test("Standardization with sparse input") {
-    val data = Array(
-      Vectors.sparse(3, Seq((0, -2.0), (1, 2.3))),
-      Vectors.sparse(3, Seq((1, -1.0), (2, -3.0))),
-      Vectors.sparse(3, Seq((1, -5.1))),
-      Vectors.sparse(3, Seq((0, 3.8), (2, 1.9))),
-      Vectors.sparse(3, Seq((0, 1.7), (1, -0.6))),
-      Vectors.sparse(3, Seq((1, 1.9)))
-    )
 
-    val dataRDD = sc.parallelize(data, 3)
+    val dataRDD = sc.parallelize(sparseData, 3)
 
     val standardizer1 = new StandardScaler(withMean = true, withStd = true)
     val standardizer2 = new StandardScaler()
@@ -127,49 +242,52 @@ class StandardScalerSuite extends FunSuite with MLlibTestSparkContext {
     val model2 = standardizer2.fit(dataRDD)
     val model3 = standardizer3.fit(dataRDD)
 
-    val data2 = data.map(model2.transform)
+    val data2 = sparseData.map(model2.transform)
 
     withClue("Standardization with mean can not be applied on sparse input.") {
       intercept[IllegalArgumentException] {
-        data.map(model1.transform)
+        sparseData.map(model1.transform)
       }
     }
 
     withClue("Standardization with mean can not be applied on sparse input.") {
       intercept[IllegalArgumentException] {
-        data.map(model3.transform)
+        sparseData.map(model3.transform)
       }
     }
 
     val data2RDD = model2.transform(dataRDD)
 
-    val summary2 = computeSummary(data2RDD)
+    validateSparse(data2, data2RDD)
+  }
 
-    assert((data, data2, data2RDD.collect()).zipped.forall {
-      case (v1: DenseVector, v2: DenseVector, v3: DenseVector) => true
-      case (v1: SparseVector, v2: SparseVector, v3: SparseVector) => true
-      case _ => false
-    }, "The vector type should be preserved after standardization.")
+  test("Standardization with constant input when means and variances are provided") {
 
-    assert((data2, data2RDD.collect()).zipped.forall((v1, v2) => v1 ~== v2 absTol 1E-5))
+    val dataRDD = sc.parallelize(constantData, 2)
 
-    assert(summary2.mean !~== Vectors.dense(0.0, 0.0, 0.0) absTol 1E-5)
-    assert(summary2.variance ~== Vectors.dense(1.0, 1.0, 1.0) absTol 1E-5)
+    val standardizer1 = new StandardScaler(withMean = true, withStd = true)
+    val standardizer2 = new StandardScaler(withMean = true, withStd = false)
+    val standardizer3 = new StandardScaler(withMean = false, withStd = true)
+
+    val model1 = standardizer1.fit(dataRDD)
+    val model2 = standardizer2.fit(dataRDD)
+    val model3 = standardizer3.fit(dataRDD)
+
+    val equivalentModel1 = new StandardScalerModel(model1.mean, model1.variance, true, true)
+    val equivalentModel2 = new StandardScalerModel(model2.mean, model2.variance, true, false)
+    val equivalentModel3 = new StandardScalerModel(model3.mean, model3.variance, false, true)
+
+    val data1 = constantData.map(equivalentModel1.transform)
+    val data2 = constantData.map(equivalentModel2.transform)
+    val data3 = constantData.map(equivalentModel3.transform)
+
+    validateConstant(data1, data2, data3)
 
-    assert(data2(4) ~== Vectors.sparse(3, Seq((0, 0.865538862), (1, -0.22604255))) absTol 1E-5)
-    assert(data2(5) ~== Vectors.sparse(3, Seq((1, 0.71580142))) absTol 1E-5)
   }
 
   test("Standardization with constant input") {
-    // When the input data is all constant, the variance is zero. The standardization against
-    // zero variance is not well-defined, but we decide to just set it into zero here.
-    val data = Array(
-      Vectors.dense(2.0),
-      Vectors.dense(2.0),
-      Vectors.dense(2.0)
-    )
 
-    val dataRDD = sc.parallelize(data, 2)
+    val dataRDD = sc.parallelize(constantData, 2)
 
     val standardizer1 = new StandardScaler(withMean = true, withStd = true)
     val standardizer2 = new StandardScaler(withMean = true, withStd = false)
@@ -179,16 +297,12 @@ class StandardScalerSuite extends FunSuite with MLlibTestSparkContext {
     val model2 = standardizer2.fit(dataRDD)
     val model3 = standardizer3.fit(dataRDD)
 
-    val data1 = data.map(model1.transform)
-    val data2 = data.map(model2.transform)
-    val data3 = data.map(model3.transform)
+    val data1 = constantData.map(model1.transform)
+    val data2 = constantData.map(model2.transform)
+    val data3 = constantData.map(model3.transform)
+
+    validateConstant(data1, data2, data3)
 
-    assert(data1.forall(_.toArray.forall(_ == 0.0)),
-      "The variance is zero, so the transformed result should be 0.0")
-    assert(data2.forall(_.toArray.forall(_ == 0.0)),
-      "The variance is zero, so the transformed result should be 0.0")
-    assert(data3.forall(_.toArray.forall(_ == 0.0)),
-      "The variance is zero, so the transformed result should be 0.0")
   }
 
 }