Added additional train() method to companion object for cluster count…

… and tolerance parameters.

Modified cluster initialization strategy to use an initial covariance matrix derived from the sample points used to initialize the mean.

mllib/src/main/scala/org/apache/spark/mllib/clustering/GMMExpectationMaximization.scala

@@ -32,7 +32,7 @@ object GMMExpectationMaximization {
   /**
    * Trains a GMM using the given parameters
    * 
-   * @param data training points stores as RDD[Vector]
+   * @param data training points stored as RDD[Vector]
    * @param k the number of Gaussians in the mixture
    * @param maxIterations the maximum number of iterations to perform
    * @param delta change in log-likelihood at which convergence is considered achieved
@@ -47,7 +47,7 @@ object GMMExpectationMaximization {
   /**
    * Trains a GMM using the given parameters
    * 
-   * @param data training points stores as RDD[Vector]
+   * @param data training points stored as RDD[Vector]
    * @param k the number of Gaussians in the mixture
    * @param maxIterations the maximum number of iterations to perform
    */
@@ -58,7 +58,18 @@ object GMMExpectationMaximization {
   /**
    * Trains a GMM using the given parameters
    * 
-   * @param data training points stores as RDD[Vector]
+   * @param data training points stored as RDD[Vector]
+   * @param k the number of Gaussians in the mixture
+   * @param delta change in log-likelihood at which convergence is considered achieved
+   */
+  def train(data: RDD[Vector], k: Int, delta: Double): GaussianMixtureModel = {
+    new GMMExpectationMaximization().setK(k).setDelta(delta).run(data)
+  }
+
+  /**
+   * Trains a GMM using the given parameters
+   * 
+   * @param data training points stored as RDD[Vector]
    * @param k the number of Gaussians in the mixture
    */
   def train(data: RDD[Vector], k: Int): GaussianMixtureModel = {
@@ -127,10 +138,12 @@ class GMMExpectationMaximization private (
 
     // C will be array of (weight, mean, covariance) tuples
     // we start with uniform weights, a random mean from the data, and
-    // identity matrices for covariance 
+    // diagonal covariance matrices using component variances
+    // derived from the samples 
     var C = (0 until k).map(i => (1.0/k, 
                                   vec_mean(samples.slice(i * nSamples, (i + 1) * nSamples)), 
-                                  BreezeMatrix.eye[Double](d))).toArray
+                                  init_cov(samples.slice(i * nSamples, (i + 1) * nSamples)))
+                           ).toArray
 
     val acc_w     = new Array[Accumulator[Double]](k)
     val acc_mu    = new Array[Accumulator[DenseDoubleVector]](k)
@@ -216,6 +229,19 @@ class GMMExpectationMaximization private (
     v / x.length.asInstanceOf[Double] 
   }
 
+  /**
+   * Construct matrix where diagonal entries are element-wise
+   * variance of input vectors (computes biased variance)
+   */
+  private def init_cov(x : Array[DenseDoubleVector]) : DenseDoubleMatrix = {
+    val mu = vec_mean(x)
+    val ss = BreezeVector.zeros[Double](x(0).length)
+    val result = BreezeMatrix.eye[Double](ss.length)
+    (0 until x.length).map(i => (x(i) - mu) :^ 2.0).foreach(u => ss += u)
+    (0 until ss.length).foreach(i => result(i,i) = ss(i) / x.length)
+    result
+  }
+
   /** AccumulatorParam for Dense Breeze Vectors */
   private object DenseDoubleVectorAccumulatorParam extends AccumulatorParam[DenseDoubleVector] {
     def zero(initialVector : DenseDoubleVector) : DenseDoubleVector = {