init pr

Author: WeichenXu123

mllib/src/main/scala/org/apache/spark/ml/stat/Summarizer.scala

@@ -438,6 +438,10 @@ private[ml] object SummaryBuilderImpl extends Logging {
         while (i < len) {
           realVariance(i) = (currM2n(i) + deltaMean(i) * deltaMean(i) * weightSum(i) *
             (totalWeightSum - weightSum(i)) / totalWeightSum) / denominator
+          // Because of numerical error, it is possible to get negative real variance
+          if (realVariance(i) < 0.0) {
+            realVariance(i) = 0.0
+          }
           i += 1
         }
       }

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

@@ -215,6 +215,10 @@ class MultivariateOnlineSummarizer extends MultivariateStatisticalSummary with S
       while (i < len) {
         realVariance(i) = (currM2n(i) + deltaMean(i) * deltaMean(i) * weightSum(i) *
           (totalWeightSum - weightSum(i)) / totalWeightSum) / denominator
+        // Because of numerical error, it is possible to get negative real variance
+        if (realVariance(i) < 0.0) {
+          realVariance(i) = 0.0
+        }
         i += 1
       }
     }

mllib/src/test/scala/org/apache/spark/ml/stat/SummarizerSuite.scala

@@ -402,6 +402,24 @@ class SummarizerSuite extends SparkFunSuite with MLlibTestSparkContext {
     assert(summarizer.count === 6)
   }
 
+  test("summarizer buffer zero variance test (SPARK-21818)") {
+    val summarizer1 = new SummarizerBuffer()
+      .add(Vectors.dense(3.0), 0.7)
+    val summarizer2 = new SummarizerBuffer()
+      .add(Vectors.dense(3.0), 0.4)
+    val summarizer3 = new SummarizerBuffer()
+      .add(Vectors.dense(3.0), 0.5)
+    val summarizer4 = new SummarizerBuffer()
+      .add(Vectors.dense(3.0), 0.4)
+
+    val summarizer = summarizer1
+      .merge(summarizer2)
+      .merge(summarizer3)
+      .merge(summarizer4)
+
+    assert(summarizer.variance(0) >= 0.0)
+  }
+
   test("summarizer buffer merging summarizer with empty summarizer") {
     // If one of two is non-empty, this should return the non-empty summarizer.
     // If both of them are empty, then just return the empty summarizer.

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

@@ -270,4 +270,22 @@ class MultivariateOnlineSummarizerSuite extends SparkFunSuite {
     assert(summarizer3.max ~== Vectors.dense(10.0, 0.0) absTol 1e-14)
     assert(summarizer3.min ~== Vectors.dense(0.0, -10.0) absTol 1e-14)
   }
+
+  test ("test zero variance (SPARK-21818)") {
+    val summarizer1 = (new MultivariateOnlineSummarizer)
+      .add(Vectors.dense(3.0), 0.7)
+    val summarizer2 = (new MultivariateOnlineSummarizer)
+      .add(Vectors.dense(3.0), 0.4)
+    val summarizer3 = (new MultivariateOnlineSummarizer)
+      .add(Vectors.dense(3.0), 0.5)
+    val summarizer4 = (new MultivariateOnlineSummarizer)
+      .add(Vectors.dense(3.0), 0.4)
+
+    val summarizer = summarizer1
+      .merge(summarizer2)
+      .merge(summarizer3)
+      .merge(summarizer4)
+
+    assert(summarizer.variance(0) >= 0.0)
+  }
 }