convert rdd into RDD of Vector

python/pyspark/mllib/stat.py

@@ -22,7 +22,7 @@
 from functools import wraps
 
 from pyspark import PickleSerializer
-from pyspark.mllib.linalg import _to_java_object_rdd
+from pyspark.mllib.linalg import _convert_to_vector, _to_java_object_rdd
 
 
 __all__ = ['MultivariateStatisticalSummary', 'Statistics']
@@ -107,7 +107,7 @@ def colStats(rdd):
         array([ 2.,  0.,  0., -2.])
         """
         sc = rdd.ctx
-        jrdd = _to_java_object_rdd(rdd)
+        jrdd = _to_java_object_rdd(rdd.map(_convert_to_vector))
         cStats = sc._jvm.PythonMLLibAPI().colStats(jrdd)
         return MultivariateStatisticalSummary(sc, cStats)
 
@@ -163,13 +163,14 @@ def corr(x, y=None, method=None):
         if type(y) == str:
             raise TypeError("Use 'method=' to specify method name.")
 
-        jx = _to_java_object_rdd(x)
         if not y:
+            jx = _to_java_object_rdd(x.map(_convert_to_vector))
             resultMat = sc._jvm.PythonMLLibAPI().corr(jx, method)
             bytes = sc._jvm.SerDe.dumps(resultMat)
             ser = PickleSerializer()
             return ser.loads(str(bytes)).toArray()
         else:
+            jx = _to_java_object_rdd(x)
             jy = _to_java_object_rdd(y)
             return sc._jvm.PythonMLLibAPI().corr(jx, jy, method)
 

python/pyspark/mllib/tests.py

@@ -36,6 +36,8 @@
 from pyspark.serializers import PickleSerializer
 from pyspark.mllib.linalg import Vector, SparseVector, DenseVector, _convert_to_vector
 from pyspark.mllib.regression import LabeledPoint
+from pyspark.mllib.random import RandomRDDs
+from pyspark.mllib.stat import Statistics
 from pyspark.tests import ReusedPySparkTestCase as PySparkTestCase
 
 
@@ -202,6 +204,16 @@ def test_regression(self):
         self.assertTrue(dt_model.predict(features[3]) > 0)
 
 
+class StatTests(PySparkTestCase):
+    # SPARK-4023
+    def test_col_with_random_rdd(self):
+        data = RandomRDDs.normalVectorRDD(self.sc, 1000, 10, 10)
+        summary = Statistics.colStats(data)
+        self.assertEqual(1000, summary.count())
+        mean = summary.mean()
+        self.assertTrue(all(abs(v) < 0.1 for v in mean))
+
+
 @unittest.skipIf(not _have_scipy, "SciPy not installed")
 class SciPyTests(PySparkTestCase):