Address comments.

Author: yanboliang

mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala

@@ -347,6 +347,7 @@ class GaussianMixture @Since("2.0.0") (
 
     val instr = Instrumentation.create(this, instances)
     instr.logParams(featuresCol, predictionCol, probabilityCol, k, maxIter, seed, tol)
+    instr.logNumFeatures(numFeatures)
 
     val shouldDistributeGaussians = GaussianMixture.shouldDistributeGaussians(
       numClusters, numFeatures)
@@ -409,16 +410,14 @@ class GaussianMixture @Since("2.0.0") (
     }
 
     val gaussianDists = gaussians.map { case (mean, covVec) =>
-      val cov = new DenseMatrix(numFeatures, numFeatures,
-        GaussianMixture.unpackUpperTriangularMatrix(numFeatures, covVec.values))
+      val cov = GaussianMixture.unpackUpperTriangularMatrix(numFeatures, covVec.values)
       new MultivariateGaussian(mean, cov)
     }
 
     val model = copyValues(new GaussianMixtureModel(uid, weights, gaussianDists)).setParent(this)
     val summary = new GaussianMixtureSummary(model.transform(dataset),
       $(predictionCol), $(probabilityCol), $(featuresCol), $(k))
     model.setSummary(Some(summary))
-    instr.logNumFeatures(model.gaussians.head.mean.size)
     instr.logSuccess(model)
     model
   }
@@ -439,7 +438,7 @@ class GaussianMixture @Since("2.0.0") (
    * @param numFeatures The number of features of training instance.
    * @return The initialized weights and corresponding gaussian distributions. Note the
    *         covariance matrix of multivariate gaussian distribution is symmetric and
-   *         we only save the upper triangular part as a dense vector.
+   *         we only save the upper triangular part as a dense vector (column major).
    */
   private def initRandom(
       instances: RDD[Vector],
@@ -463,8 +462,8 @@ class GaussianMixture @Since("2.0.0") (
          Construct matrix where diagonal entries are element-wise
          variance of input vectors (computes biased variance).
          Since the covariance matrix of multivariate gaussian distribution is symmetric,
-         only the upper triangular part of the matrix will be saved as a dense vector
-         in order to reduce the shuffled data size.
+         only the upper triangular part of the matrix (column major) will be saved as
+         a dense vector in order to reduce the shuffled data size.
        */
       val cov = {
         val ss = new DenseVector(new Array[Double](numFeatures)).asBreeze
@@ -505,20 +504,20 @@ object GaussianMixture extends DefaultParamsReadable[GaussianMixture] {
 
   /**
    * Convert an n * (n + 1) / 2 dimension array representing the upper triangular part of a matrix
-   * into an n * n array representing the full symmetric matrix.
+   * into an n * n array representing the full symmetric matrix (column major).
    *
    * @param n The order of the n by n matrix.
    * @param triangularValues The upper triangular part of the matrix packed in an array
    *                         (column major).
-   * @return An array which represents the symmetric matrix in column major.
+   * @return A dense matrix which represents the symmetric matrix in column major.
    */
   private[clustering] def unpackUpperTriangularMatrix(
       n: Int,
-      triangularValues: Array[Double]): Array[Double] = {
+      triangularValues: Array[Double]): DenseMatrix = {
     val symmetricValues = new Array[Double](n * n)
     var r = 0
     var i = 0
-    while(i < n) {
+    while (i < n) {
       var j = 0
       while (j <= i) {
         symmetricValues(i * n + j) = triangularValues(r)
@@ -528,15 +527,15 @@ object GaussianMixture extends DefaultParamsReadable[GaussianMixture] {
       }
       i += 1
     }
-    symmetricValues
+    new DenseMatrix(n, n, symmetricValues)
   }
 
   /**
    * Update the weight, mean and covariance of gaussian distribution.
    *
    * @param mean The mean of the gaussian distribution.
    * @param cov The covariance matrix of the gaussian distribution. Note we only
-   *            save the upper triangular part as a dense vector.
+   *            save the upper triangular part as a dense vector (column major).
    * @param weight The weight of the gaussian distribution.
    * @param sumWeights The sum of weights of all clusters.
    * @return The updated weight, mean and covariance.
@@ -562,8 +561,8 @@ object GaussianMixture extends DefaultParamsReadable[GaussianMixture] {
  * @param bcWeights The broadcast weights for each Gaussian distribution in the mixture.
  * @param bcGaussians The broadcast array of Multivariate Gaussian (Normal) Distribution
  *                    in the mixture. Note only upper triangular part of the covariance
- *                    matrix of each distribution is stored as dense vector in order to
- *                    reduce shuffled data size.
+ *                    matrix of each distribution is stored as dense vector (column major)
+ *                    in order to reduce shuffled data size.
  */
 private class ExpectationAggregator(
     numFeatures: Int,
@@ -581,8 +580,7 @@ private class ExpectationAggregator(
 
   @transient private lazy val oldGaussians = {
     bcGaussians.value.map { case (mean, covVec) =>
-      val cov = new DenseMatrix(numFeatures, numFeatures,
-        GaussianMixture.unpackUpperTriangularMatrix(numFeatures, covVec.values))
+      val cov = GaussianMixture.unpackUpperTriangularMatrix(numFeatures, covVec.values)
       new MultivariateGaussian(mean, cov)
     }
   }
@@ -611,7 +609,7 @@ private class ExpectationAggregator(
     val prob = new Array[Double](k)
     var probSum = 0.0
     var i = 0
-    while(i < k) {
+    while (i < k) {
       val p = EPSILON + localWeights(i) * localOldGaussians(i).pdf(instance)
       prob(i) = p
       probSum += p
@@ -623,7 +621,7 @@ private class ExpectationAggregator(
     val localNewMeans = newMeans
     val localNewCovs = newCovs
     i = 0
-    while(i < k) {
+    while (i < k) {
       prob(i) /= probSum
       localNewWeights(i) += prob(i)
       BLAS.axpy(prob(i), instance, localNewMeans(i))
@@ -654,7 +652,7 @@ private class ExpectationAggregator(
       val localThisNewCovs = this.newCovs
       val localOtherNewCovs = other.newCovs
       var i = 0
-      while(i < k) {
+      while (i < k) {
         localThisNewWeights(i) += localOtherNewWeights(i)
         BLAS.axpy(1.0, localOtherNewMeans(i), localThisNewMeans(i))
         BLAS.axpy(1.0, localOtherNewCovs(i), localThisNewCovs(i))

mllib/src/test/scala/org/apache/spark/ml/clustering/GaussianMixtureSuite.scala

@@ -18,8 +18,9 @@
 package org.apache.spark.ml.clustering
 
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.ml.linalg.{Matrices, Vector, Vectors}
+import org.apache.spark.ml.linalg.{DenseMatrix, Matrices, Vector, Vectors}
 import org.apache.spark.ml.param.ParamMap
+import org.apache.spark.ml.stat.distribution.MultivariateGaussian
 import org.apache.spark.ml.util.{DefaultReadWriteTest, MLTestingUtils}
 import org.apache.spark.ml.util.TestingUtils._
 import org.apache.spark.mllib.util.MLlibTestSparkContext
@@ -33,6 +34,7 @@ class GaussianMixtureSuite extends SparkFunSuite with MLlibTestSparkContext
   import GaussianMixtureSuite._
 
   final val k = 5
+  private val seed = 538009335
   @transient var dataset: Dataset[_] = _
   @transient var denseDataset: Dataset[_] = _
   @transient var sparseDataset: Dataset[_] = _
@@ -45,7 +47,7 @@ class GaussianMixtureSuite extends SparkFunSuite with MLlibTestSparkContext
     dataset = KMeansSuite.generateKMeansData(spark, 50, 3, k)
     denseDataset = denseData.map(FeatureData).toDF()
     sparseDataset = denseData.map { point =>
-      FeatureData(Vectors.sparse(1, Array(0), point.toArray))
+      FeatureData(point.toSparse)
     }.toDF()
     decompositionDataset = decompositionData.map(FeatureData).toDF()
     rDataset = rData.map(FeatureData).toDF()
@@ -144,40 +146,36 @@ class GaussianMixtureSuite extends SparkFunSuite with MLlibTestSparkContext
 
   test("univariate dense data with two clusters") {
     val weights = Array(2.0 / 3.0, 1.0 / 3.0)
-    val mean = Array(Vectors.dense(5.1604), Vectors.dense(-4.3673))
-    val cov = Array(Matrices.dense(1, 1, Array(0.86644)), Matrices.dense(1, 1, Array(1.1098)))
-
-    val gmm = new GaussianMixture().setK(2).fit(denseDataset)
+    val means = Array(Vectors.dense(5.1604), Vectors.dense(-4.3673))
+    val covs = Array(Matrices.dense(1, 1, Array(0.86644)), Matrices.dense(1, 1, Array(1.1098)))
+    val gaussians = means.zip(covs).map { case (mean, cov) =>
+      new MultivariateGaussian(mean, cov)
+    }
 
-    assert(gmm.weights(0) ~== weights(0) absTol 1E-3)
-    assert(gmm.weights(1) ~== weights(1) absTol 1E-3)
-    assert(gmm.gaussians(0).mean ~== mean(0) absTol 1E-3)
-    assert(gmm.gaussians(1).mean ~== mean(1) absTol 1E-3)
-    assert(gmm.gaussians(0).cov ~== cov(0) absTol 1E-3)
-    assert(gmm.gaussians(1).cov ~== cov(1) absTol 1E-3)
+    val expected = new GaussianMixtureModel("dummy", weights, gaussians)
+    val actual = new GaussianMixture().setK(2).setSeed(seed).fit(denseDataset)
+    modelEquals(expected, actual)
   }
 
   test("univariate sparse data with two clusters") {
     val weights = Array(2.0 / 3.0, 1.0 / 3.0)
-    val mean = Array(Vectors.dense(5.1604), Vectors.dense(-4.3673))
-    val cov = Array(Matrices.dense(1, 1, Array(0.86644)), Matrices.dense(1, 1, Array(1.1098)))
-
-    val gmm = new GaussianMixture().setK(2).fit(sparseDataset)
+    val means = Array(Vectors.dense(5.1604), Vectors.dense(-4.3673))
+    val covs = Array(Matrices.dense(1, 1, Array(0.86644)), Matrices.dense(1, 1, Array(1.1098)))
+    val gaussians = means.zip(covs).map { case (mean, cov) =>
+      new MultivariateGaussian(mean, cov)
+    }
 
-    assert(gmm.weights(0) ~== weights(0) absTol 1E-3)
-    assert(gmm.weights(1) ~== weights(1) absTol 1E-3)
-    assert(gmm.gaussians(0).mean ~== mean(0) absTol 1E-3)
-    assert(gmm.gaussians(1).mean ~== mean(1) absTol 1E-3)
-    assert(gmm.gaussians(0).cov ~== cov(0) absTol 1E-3)
-    assert(gmm.gaussians(1).cov ~== cov(1) absTol 1E-3)
+    val expected = new GaussianMixtureModel("dummy", weights, gaussians)
+    val actual = new GaussianMixture().setK(2).setSeed(seed).fit(sparseDataset)
+    modelEquals(expected, actual)
   }
 
   test("check distributed decomposition") {
     val k = 5
     val d = decompositionData.head.size
     assert(GaussianMixture.shouldDistributeGaussians(k, d))
 
-    val gmm = new GaussianMixture().setK(k).fit(decompositionDataset)
+    val gmm = new GaussianMixture().setK(k).setSeed(seed).fit(decompositionDataset)
     assert(gmm.getK === k)
   }
 
@@ -213,18 +211,16 @@ class GaussianMixtureSuite extends SparkFunSuite with MLlibTestSparkContext
       [2,] 0.1607830 1.008878
      */
     val weights = Array(0.5333333, 0.4666667)
-    val mean = Array(Vectors.dense(10.363673, 9.897081), Vectors.dense(0.11731091, -0.06192351))
-    val cov = Array(Matrices.dense(2, 2, Array(0.2961543, 0.1607830, 0.160783, 1.008878)),
+    val means = Array(Vectors.dense(10.363673, 9.897081), Vectors.dense(0.11731091, -0.06192351))
+    val covs = Array(Matrices.dense(2, 2, Array(0.2961543, 0.1607830, 0.160783, 1.008878)),
       Matrices.dense(2, 2, Array(0.62049934, 0.06880802, 0.06880802, 1.27431874)))
+    val gaussians = means.zip(covs).map { case (mean, cov) =>
+      new MultivariateGaussian(mean, cov)
+    }
 
-    val gmm = new GaussianMixture().setK(2).fit(rDataset)
-
-    assert(gmm.weights(0) ~== weights(0) absTol 1E-3)
-    assert(gmm.weights(1) ~== weights(1) absTol 1E-3)
-    assert(gmm.gaussians(0).mean ~== mean(0) absTol 1E-3)
-    assert(gmm.gaussians(1).mean ~== mean(1) absTol 1E-3)
-    assert(gmm.gaussians(0).cov ~== cov(0) absTol 1E-3)
-    assert(gmm.gaussians(1).cov ~== cov(1) absTol 1E-3)
+    val expected = new GaussianMixtureModel("dummy", weights, gaussians)
+    val actual = new GaussianMixture().setK(2).setSeed(seed).fit(rDataset)
+    modelEquals(expected, actual)
   }
 
   test("upper triangular matrix unpacking") {
@@ -238,12 +234,13 @@ class GaussianMixtureSuite extends SparkFunSuite with MLlibTestSparkContext
     val triangularValues = Array(1.0, 2.5, 2.0, 3.8, 7.2, 3.0, 0.9, 3.8, 1.0, 4.0)
     val symmetricValues = Array(1.0, 2.5, 3.8, 0.9, 2.5, 2.0, 7.2, 3.8,
       3.8, 7.2, 3.0, 1.0, 0.9, 3.8, 1.0, 4.0)
-    val expected = GaussianMixture.unpackUpperTriangularMatrix(4, triangularValues)
-    assert(symmetricValues === expected)
+    val symmetricMatrix = new DenseMatrix(4, 4, symmetricValues)
+    val expectedMatrix = GaussianMixture.unpackUpperTriangularMatrix(4, triangularValues)
+    assert(symmetricMatrix === expectedMatrix)
   }
 }
 
-object GaussianMixtureSuite {
+object GaussianMixtureSuite extends SparkFunSuite {
   /**
    * Mapping from all Params to valid settings which differ from the defaults.
    * This is useful for tests which need to exercise all Params, such as save/load.
@@ -281,4 +278,12 @@ object GaussianMixtureSuite {
   )
 
   case class FeatureData(features: Vector)
+
+  def modelEquals(m1: GaussianMixtureModel, m2: GaussianMixtureModel): Unit = {
+    assert(m1.weights.length === m2.weights.length)
+    for (i <- m1.weights.indices) {
+      assert(m1.gaussians(i).mean ~== m2.gaussians(i).mean absTol 1E-3)
+      assert(m1.gaussians(i).cov ~== m2.gaussians(i).cov absTol 1E-3)
+    }
+  }
 }