changed to sample covariance

Author: brkyvz

python/pyspark/sql/dataframe.py

@@ -877,8 +877,8 @@ def fillna(self, value, subset=None):
 
     def cov(self, col1, col2):
         """
-        Calculate the covariance for the given columns, specified by their names as a double value.
-        :func:`DataFrame.cov` and :func:`DataFrameStatFunctions.cov` are aliases.
+        Calculate the sample covariance for the given columns, specified by their names, as a
+        double value. :func:`DataFrame.cov` and :func:`DataFrameStatFunctions.cov` are aliases.
 
         :param col1: The name of the first column
         :param col2: The name of the second column

python/pyspark/sql/tests.py

@@ -390,7 +390,7 @@ def test_aggregator(self):
     def test_cov(self):
         df = self.sc.parallelize([Row(a=i, b=2 * i) for i in range(10)]).toDF()
         cov = df.stat.cov("a", "b")
-        self.assertTrue(abs(cov - 16.5) < 1e-6)
+        self.assertTrue(abs(cov - 55.0 / 3) < 1e-6)
 
     def test_math_functions(self):
         df = self.sc.parallelize([Row(a=i, b=2 * i) for i in range(10)]).toDF()

sql/core/src/main/scala/org/apache/spark/sql/DataFrameStatFunctions.scala

@@ -67,7 +67,7 @@ final class DataFrameStatFunctions private[sql](df: DataFrame) {
   }
 
   /**
-   * Calculate the covariance of two numerical columns of a DataFrame.
+   * Calculate the sample covariance of two numerical columns of a DataFrame.
    * @param col1 the name of the first column
    * @param col2 the name of the second column
    * @return the covariance of the two columns.

sql/core/src/main/scala/org/apache/spark/sql/execution/stat/StatFunctions.scala

@@ -49,8 +49,8 @@ private[sql] object StatFunctions {
       count = totalCount
       this
     }
-    // return the covariance for the observed examples
-    def cov: Double = Ck / count
+    // return the sample covariance for the observed examples
+    def cov: Double = Ck / (count - 1)
   }
 
   /**

sql/core/src/test/scala/org/apache/spark/sql/DataFrameStatSuite.scala

@@ -51,7 +51,7 @@ class DataFrameStatSuite extends FunSuite  {
     val df = sqlCtx.sparkContext.parallelize(rows).toDF("singles", "doubles", "letters")
 
     val results = df.stat.cov("singles", "doubles")
-    assert(math.abs(results - 16.5) < 1e-6)
+    assert(math.abs(results - 55.0 / 3) < 1e-6)
     intercept[IllegalArgumentException] {
       df.stat.cov("singles", "letters") // doesn't accept non-numerical dataTypes
     }