[SPARK-7844] Fix broken tests in KernelDensity

The densities in KernelDensity are scaled down by
(number of parallel processes X number of points). This results in broken tests
in KernelDensitySuite which haven't been tested properly.

Author: MechCoder

mllib/src/main/scala/org/apache/spark/mllib/stat/KernelDensity.scala

@@ -86,20 +86,20 @@ class KernelDensity extends Serializable {
     val n = points.length
     // This gets used in each Gaussian PDF computation, so compute it up front
     val logStandardDeviationPlusHalfLog2Pi = math.log(bandwidth) + 0.5 * math.log(2 * math.Pi)
-    val (densities, count) = sample.aggregate((new Array[Double](n), 0L))(
+    val densities = sample.aggregate(new Array[Double](n))(
       (x, y) => {
         var i = 0
         while (i < n) {
-          x._1(i) += normPdf(y, bandwidth, logStandardDeviationPlusHalfLog2Pi, points(i))
+          x(i) += normPdf(y, bandwidth, logStandardDeviationPlusHalfLog2Pi, points(i))
           i += 1
         }
-        (x._1, n)
+        x
       },
       (x, y) => {
-        blas.daxpy(n, 1.0, y._1, 1, x._1, 1)
-        (x._1, x._2 + y._2)
+        blas.daxpy(n, 1.0, y, 1, x, 1)
+        x
       })
-    blas.dscal(n, 1.0 / count, densities, 1)
+    blas.dscal(n, 1.0 / sample.count, densities, 1)
     densities
   }
 }

mllib/src/test/scala/org/apache/spark/mllib/stat/KernelDensitySuite.scala

@@ -29,8 +29,8 @@ class KernelDensitySuite extends FunSuite with MLlibTestSparkContext {
     val densities = new KernelDensity().setSample(rdd).setBandwidth(3.0).estimate(evaluationPoints)
     val normal = new NormalDistribution(5.0, 3.0)
     val acceptableErr = 1e-6
-    assert(densities(0) - normal.density(5.0) < acceptableErr)
-    assert(densities(0) - normal.density(6.0) < acceptableErr)
+    assert(Math.abs(densities(0) - normal.density(5.0)) < acceptableErr)
+    assert(Math.abs(densities(1) - normal.density(6.0)) < acceptableErr)
   }
 
   test("kernel density multiple samples") {
@@ -40,7 +40,9 @@ class KernelDensitySuite extends FunSuite with MLlibTestSparkContext {
     val normal1 = new NormalDistribution(5.0, 3.0)
     val normal2 = new NormalDistribution(10.0, 3.0)
     val acceptableErr = 1e-6
-    assert(densities(0) - (normal1.density(5.0) + normal2.density(5.0)) / 2 < acceptableErr)
-    assert(densities(0) - (normal1.density(6.0) + normal2.density(6.0)) / 2 < acceptableErr)
+    assert(Math.abs(
+      densities(0) - (normal1.density(5.0) + normal2.density(5.0)) / 2) < acceptableErr)
+    assert(Math.abs(
+      densities(1) - (normal1.density(6.0) + normal2.density(6.0)) / 2) < acceptableErr)
   }
 }