Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-4023] [MLlib] [PySpark] convert rdd into RDD of Vector #2870

Closed
wants to merge 2 commits into from
Closed

[SPARK-4023] [MLlib] [PySpark] convert rdd into RDD of Vector #2870

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0871576
convert rdd into RDD of Vector
Oct 21, 2014
1eac767
address comments
Oct 21, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions python/pyspark/mllib/stat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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']
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -163,14 +163,15 @@ 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:
jy = _to_java_object_rdd(y)
jx = _to_java_object_rdd(x.map(float))
jy = _to_java_object_rdd(y.map(float))
return sc._jvm.PythonMLLibAPI().corr(jx, jy, method)


Expand Down
19 changes: 19 additions & 0 deletions python/pyspark/mllib/tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -202,6 +204,23 @@ def test_regression(self):
self.assertTrue(dt_model.predict(features[3]) > 0)


class StatTests(PySparkTestCase):
# SPARK-4023
def test_col_with_different_rdds(self):
# numpy
data = RandomRDDs.normalVectorRDD(self.sc, 1000, 10, 10)
summary = Statistics.colStats(data)
self.assertEqual(1000, summary.count())
# array
data = self.sc.parallelize([range(10)] * 10)
summary = Statistics.colStats(data)
self.assertEqual(10, summary.count())
# array
data = self.sc.parallelize([pyarray.array("d", range(10))] * 10)
summary = Statistics.colStats(data)
self.assertEqual(10, summary.count())


@unittest.skipIf(not _have_scipy, "SciPy not installed")
class SciPyTests(PySparkTestCase):

Expand Down