[SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans #6499

MechCoder · 2015-05-29T13:32:42Z

Python bindings for StreamingKMeans

Will change status to MRG once docs, tests and examples are updated.

SparkQA · 2015-05-29T13:41:31Z

Test build #33738 has finished for PR 6499 at commit 773c555.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

mengxr · 2015-05-29T18:15:24Z

cc @freeman-lab

MechCoder · 2015-05-29T20:03:16Z

It's still in WIP yet.

MechCoder · 2015-05-30T13:39:23Z

@mengxr @freeman-lab

I would really like some help here. The StreamingKMeansModel works as it is supposed to. I have added doc tests for those.

However, the trainOn method of StreamingKMeans (https://github.com/apache/spark/pull/6499/files#diff-21c55b407050d37f67a2919470e047ebR373) seems to run indefinitely. I printed 'rdd.collect()' under if rdd, it seems the foreachRDD continues even after all the batches in the dstream are exhausted.

A small example to reproduce.

from pyspark.mllib.clustering import StreamingKMeans, StreamingKMeansModel
stk = StreamingKMeans()
initCenters = [[0.0, 0.0], [1.0, 1.0]] 
weights = [1.0, 1.0]
stk.setInitialCenters(initCenters, weights)
dv1, dv2, dv3, dv4 = [-0.1, -0.1], [0.1, 0.1], [1.1, 1.1], [0.9, 0.9]
dvc = [[dv1, dv3], [dv2, dv4]]
dvc = [sc.parallelize(i, 1) for i in dvc]
ssc = StreamingContext(sc, 2.0)
input_stream = ssc.queueStream(dvc)
stk.trainOn(input_stream)
ssc.start()

This does not seem to terminate.

SparkQA · 2015-05-30T15:00:31Z

Test build #33807 has finished for PR 6499 at commit 1d8c66a.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-05-31T05:41:10Z

I narrowed it down to this small test case.

from pyspark.streaming import StreamingContext
dvc = [[-0.1, -0.1], [0.1, 0.1], [1.1, 1.1], [0.9, 0.9]]
dvc = [sc.parallelize(i, 1) for i in dvc]
ssc = StreamingContext(sc, 2.0)
input_stream = ssc.queueStream(dvc)

def get_output(rdd):
    print(rdd.collect())
input_stream.foreachRDD(get_output)
ssc.start()

Why does this run indefinitely?

MechCoder · 2015-06-01T18:33:10Z

Figured it out thanks to @davies through mail. Will finish this up tomorrow.

freeman-lab · 2015-06-02T05:55:11Z

Cool, excited to look at this! Can definitely take a pass after you update.

MechCoder · 2015-06-02T14:16:27Z

@freeman-lab @mengxr
Please give a first pass. I've update the PR with a couple of tests.
Also, Jenkins is showing me this error in Python 3 for the doctests. I wish I could understand what's going on, but for that I would need to set up a Python 3 env.
TypeError: must be a unicode character, not bytes, related to this JIRA https://issues.apache.org/jira/browse/SPARK-7379

SparkQA · 2015-06-02T14:21:22Z

Test build #33985 has finished for PR 6499 at commit 7cc2d07.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-02T16:08:09Z

Test build #33987 has finished for PR 6499 at commit eaace71.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-06-02T18:34:46Z

python/pyspark/mllib/tests.py

@@ -818,6 +830,78 @@ def test_model_transform(self):
        self.assertEqual(model.transform([1.0, 2.0, 3.0]), DenseVector([1.0, 2.0, 3.0]))


+class StreamingKMeansTest(MLLibStreamingTestCase):
+    def test_model_params(self):


@davies Could you please verify if these are the best way to add MLlib Streaming tests? I copied the idea from the tests in streaming.py.

SparkQA · 2015-06-03T18:13:26Z

Test build #34090 has finished for PR 6499 at commit 538a7ee.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-03T18:14:20Z

Test build #34089 has finished for PR 6499 at commit 756b4d8.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-04T06:35:41Z

Test build #34156 has finished for PR 6499 at commit e12c590.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-04T08:31:43Z

Test build #34158 has finished for PR 6499 at commit 2ffed68.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-04T17:01:30Z

Test build #34190 has finished for PR 6499 at commit 33351ca.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-04T18:44:27Z

Test build #34196 has finished for PR 6499 at commit 2ffed68.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-04T21:32:17Z

Test build #34206 has finished for PR 6499 at commit d37c473.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-06-05T04:08:48Z

jenkins retest this please

SparkQA · 2015-06-05T05:43:03Z

Test build #34251 has finished for PR 6499 at commit d37c473.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

mengxr · 2015-06-18T22:14:22Z

docs/mllib-clustering.md

+
+{% highlight python %}
+model.trainOn(trainingData)
+model.predictOnValues(testData.map(lambda lp: (lp.label, lp.features)))


There is no output from this example. Call print.

SparkQA · 2015-06-18T22:16:43Z

Test build #35161 has finished for PR 6499 at commit ff0a1fc.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-18T22:18:20Z

Test build #35163 has finished for PR 6499 at commit 2061a76.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-06-19T06:44:56Z

@mengxr updated ! please review.

SparkQA · 2015-06-19T07:30:26Z

Test build #35232 has finished for PR 6499 at commit 6e87635.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-06-19T07:35:16Z

jenkins retest this please.

SparkQA · 2015-06-19T08:19:51Z

Test build #35235 has finished for PR 6499 at commit 51052d3.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-19T08:44:50Z

Test build #35238 has finished for PR 6499 at commit 7722d16.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

SparkQA · 2015-06-19T09:13:02Z

Test build #35246 has finished for PR 6499 at commit 7722d16.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

MechCoder · 2015-06-19T13:58:32Z

jenkins retest this please.

SparkQA · 2015-06-19T15:55:20Z

Test build #35269 has finished for PR 6499 at commit 7722d16.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

mengxr · 2015-06-19T17:21:37Z

test this please

mengxr · 2015-06-19T17:22:06Z

(run another time in case the streaming tests are flaky)

mengxr · 2015-06-19T17:22:14Z

LGTM pending Jenkins.

SparkQA · 2015-06-19T19:16:24Z

Test build #35289 has finished for PR 6499 at commit 7722d16.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class StreamingKMeansModel(KMeansModel):
- class StreamingKMeans(object):

mengxr · 2015-06-19T19:23:26Z

Merged into master. Thanks!

MechCoder · 2015-06-19T19:33:51Z

Thanks for your reviews and help.

MechCoder mentioned this pull request May 29, 2015

[SPARK-7946] [MLlib] DecayFactor wrongly set in StreamingKMeans #6497

Closed

MechCoder force-pushed the spark-4118 branch from 773c555 to 1d8c66a Compare May 30, 2015 13:24

MechCoder changed the title ~~[WIP] [SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans~~ [SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans May 30, 2015

MechCoder force-pushed the spark-4118 branch from cd1151e to 7cc2d07 Compare June 2, 2015 14:14

MechCoder force-pushed the spark-4118 branch from 7cc2d07 to eaace71 Compare June 2, 2015 14:24

MechCoder reviewed Jun 2, 2015
View reviewed changes

MechCoder force-pushed the spark-4118 branch from eaace71 to 756b4d8 Compare June 3, 2015 16:33

MechCoder force-pushed the spark-4118 branch from 538a7ee to e12c590 Compare June 4, 2015 04:48

MechCoder force-pushed the spark-4118 branch from 33351ca to 2ffed68 Compare June 4, 2015 17:11

MechCoder force-pushed the spark-4118 branch from d37c473 to ac10c75 Compare June 5, 2015 05:48

mengxr reviewed Jun 18, 2015
View reviewed changes

MechCoder force-pushed the spark-4118 branch from 43382e0 to 6e87635 Compare June 19, 2015 06:26

Doc fixes

51052d3

MechCoder force-pushed the spark-4118 branch from 6e87635 to 51052d3 Compare June 19, 2015 06:28

minor style fixes

7722d16

asfgit closed this in 54976e5 Jun 19, 2015

MechCoder deleted the spark-4118 branch June 19, 2015 19:26

MechCoder mentioned this pull request Jun 21, 2015

[SPARK-7104][MLlib] Support model save/load in Python's Word2Vec #6821

Closed

[SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans #6499

[SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans #6499

Conversation

MechCoder commented May 29, 2015

SparkQA commented May 29, 2015

mengxr commented May 29, 2015

MechCoder commented May 29, 2015

MechCoder commented May 30, 2015

SparkQA commented May 30, 2015

MechCoder commented May 31, 2015

MechCoder commented Jun 1, 2015

freeman-lab commented Jun 2, 2015

MechCoder commented Jun 2, 2015

SparkQA commented Jun 2, 2015

SparkQA commented Jun 2, 2015

MechCoder Jun 2, 2015

Choose a reason for hiding this comment

SparkQA commented Jun 3, 2015

SparkQA commented Jun 3, 2015

SparkQA commented Jun 4, 2015

SparkQA commented Jun 4, 2015

SparkQA commented Jun 4, 2015

SparkQA commented Jun 4, 2015

SparkQA commented Jun 4, 2015

MechCoder commented Jun 5, 2015

SparkQA commented Jun 5, 2015

mengxr Jun 18, 2015

Choose a reason for hiding this comment

SparkQA commented Jun 18, 2015

SparkQA commented Jun 18, 2015

MechCoder commented Jun 19, 2015

SparkQA commented Jun 19, 2015

MechCoder commented Jun 19, 2015

SparkQA commented Jun 19, 2015

SparkQA commented Jun 19, 2015

SparkQA commented Jun 19, 2015

MechCoder commented Jun 19, 2015

SparkQA commented Jun 19, 2015

mengxr commented Jun 19, 2015

mengxr commented Jun 19, 2015

mengxr commented Jun 19, 2015

SparkQA commented Jun 19, 2015

mengxr commented Jun 19, 2015

MechCoder commented Jun 19, 2015