{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/32751786","id":32751786,"diff_hunk":"@@ -863,6 +876,107 @@ def test_model_transform(self):\n             eprod.transform(sparsevec), SparseVector(3, [0], [3]))\n \n \n+class StreamingKMeansTest(MLLibStreamingTestCase):\n+    def test_model_params(self):\n+        stkm = StreamingKMeans()\n+        stkm.setK(5).setDecayFactor(0.0)\n+        self.assertEquals(stkm._k, 5)\n+        self.assertEquals(stkm._decayFactor, 0.0)\n+\n+        # Model not set yet.\n+        self.assertIsNone(stkm.latestModel)\n+        self.assertRaises(ValueError, stkm.trainOn, [0.0, 1.0])\n+\n+        stkm.setInitialCenters([[0.0, 0.0], [1.0, 1.0]], [1.0, 1.0])\n+        self.assertEquals(stkm.latestModel.centers, [[0.0, 0.0], [1.0, 1.0]])\n+        self.assertEquals(stkm.latestModel.getClusterWeights, [1.0, 1.0])\n+\n+    @staticmethod\n+    def _ssc_wait(start_time, end_time, sleep_time):\n+        while time() - start_time < end_time:\n+            sleep(0.01)\n+\n+    def test_accuracy_for_single_center(self):\n+        numBatches, numPoints, k, d, r, seed = 5, 5, 1, 5, 0.1, 0\n+        centers, batches = self.streamingKMeansDataGenerator(\n+            numBatches, numPoints, k, d, r, seed)\n+        stkm = StreamingKMeans(1)\n+        stkm.setInitialCenters([[0., 0., 0., 0., 0.]], [0.])\n+        input_stream = self.ssc.queueStream(\n+            [self.sc.parallelize(batch, 1) for batch in batches])\n+        stkm.trainOn(input_stream)\n+        t = time()\n+        self.ssc.start()\n+        self._ssc_wait(t, 10.0, 0.01)\n+        self.assertEquals(stkm.latestModel.getClusterWeights, [25.0])\n+        realCenters = array_sum(array(centers), axis=0)\n+        for i in range(d):\n+            modelCenters = stkm.latestModel.centers[0][i]\n+            self.assertAlmostEqual(centers[0][i], modelCenters, 1)\n+            self.assertAlmostEqual(realCenters[i], modelCenters, 1)\n+\n+    def streamingKMeansDataGenerator(self, batches, numPoints,\n+                                     k, d, r, seed, centers=None):\n+        rng = random.RandomState(seed)\n+\n+        # Generate centers.\n+        centers = [rng.randn(d) for i in range(k)]\n+\n+        return centers, [[Vectors.dense(centers[j % k] + r * rng.randn(d))\n+                          for j in range(numPoints)]\n+                         for i in range(batches)]\n+\n+    def test_trainOn_model(self):\n+        # Test the model on toy data with four clusters.\n+        stkm = StreamingKMeans()\n+        initCenters = [[1.0, 1.0], [-1.0, 1.0], [-1.0, -1.0], [1.0, -1.0]]\n+        weights = [1.0, 1.0, 1.0, 1.0]\n+        stkm.setInitialCenters(initCenters, weights)\n+\n+        # Create a toy dataset by setting a tiny offest for each point.\n+        offsets = [[0, 0.1], [0, -0.1], [0.1, 0], [-0.1, 0]]\n+        batches = []\n+        for offset in offsets:\n+            batches.append([[offset[0] + center[0], offset[1] + center[1]]\n+                            for center in initCenters])\n+\n+        batches = [self.sc.parallelize(batch, 1) for batch in batches]\n+        input_stream = self.ssc.queueStream(batches)\n+        stkm.trainOn(input_stream)\n+        t = time()\n+        self.ssc.start()\n+\n+        # Give enough time to train the model.\n+        self._ssc_wait(t, 6.0, 0.01)\n+        finalModel = stkm.latestModel\n+        self.assertTrue(all(finalModel.centers == array(initCenters)))\n+        self.assertEquals(finalModel.getClusterWeights, [5.0, 5.0, 5.0, 5.0])\n+\n+    def test_predictOn_model(self):\n+        initCenters = [[1.0, 1.0], [-1.0, 1.0], [-1.0, -1.0], [1.0, -1.0]]\n+        weights = [1.0, 1.0, 1.0, 1.0]\n+        stkm = StreamingKMeans()\n+        stkm.latestModel = StreamingKMeansModel(initCenters, weights)\n+\n+        predict_data = [[[1.5, 1.5]], [[-1.5, 1.5]], [[-1.5, -1.5]], [[1.5, -1.5]]]\n+        predict_data = [sc.parallelize(batch, 1) for batch in predict_data]\n+        predict_stream = self.ssc.queueStream(predict_data)\n+        predict_val = stkm.predictOn(predict_stream)\n+\n+        result = []\n+\n+        def update(rdd):\n+            rdd_collect = rdd.collect()\n+            if rdd_collect:\n+                result.append(rdd_collect)\n+\n+        predict_val.foreachRDD(update)\n+        t = time()\n+        self.ssc.start()\n+        self._ssc_wait(t, 6.0, 0.01)\n+        self.assertEquals(result, [[0], [1], [2], [3]])\n+\n+","path":"python/pyspark/mllib/tests.py","position":148,"original_position":148,"commit_id":"81482fd31ff597779724c2ccec7f223aa97a0f60","original_commit_id":"81482fd31ff597779724c2ccec7f223aa97a0f60","user":{"login":"freeman-lab","id":3387500,"avatar_url":"https://avatars.githubusercontent.com/u/3387500?v=3","gravatar_id":"","url":"https://api.github.com/users/freeman-lab","html_url":"https://github.com/freeman-lab","followers_url":"https://api.github.com/users/freeman-lab/followers","following_url":"https://api.github.com/users/freeman-lab/following{/other_user}","gists_url":"https://api.github.com/users/freeman-lab/gists{/gist_id}","starred_url":"https://api.github.com/users/freeman-lab/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/freeman-lab/subscriptions","organizations_url":"https://api.github.com/users/freeman-lab/orgs","repos_url":"https://api.github.com/users/freeman-lab/repos","events_url":"https://api.github.com/users/freeman-lab/events{/privacy}","received_events_url":"https://api.github.com/users/freeman-lab/received_events","type":"User","site_admin":false},"body":"Similar to this: https://github.com/apache/spark/blob/master/mllib/src/test/scala/org/apache/spark/mllib/regression/StreamingLinearRegressionSuite.scala#L142-168","created_at":"2015-06-18T16:53:47Z","updated_at":"2015-06-18T16:53:47Z","html_url":"https://github.com/apache/spark/pull/6499#discussion_r32751786","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/6499","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/32751786"},"html":{"href":"https://github.com/apache/spark/pull/6499#discussion_r32751786"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/6499"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/6499","id":36505352,"html_url":"https://github.com/apache/spark/pull/6499","diff_url":"https://github.com/apache/spark/pull/6499.diff","patch_url":"https://github.com/apache/spark/pull/6499.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/6499","number":6499,"state":"open","locked":false,"title":"[SPARK-4118] [MLlib] [PySpark] Python bindings for StreamingKMeans","user":{"login":"MechCoder","id":1867024,"avatar_url":"https://avatars.githubusercontent.com/u/1867024?v=3","gravatar_id":"","url":"https://api.github.com/users/MechCoder","html_url":"https://github.com/MechCoder","followers_url":"https://api.github.com/users/MechCoder/followers","following_url":"https://api.github.com/users/MechCoder/following{/other_user}","gists_url":"https://api.github.com/users/MechCoder/gists{/gist_id}","starred_url":"https://api.github.com/users/MechCoder/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/MechCoder/subscriptions","organizations_url":"https://api.github.com/users/MechCoder/orgs","repos_url":"https://api.github.com/users/MechCoder/repos","events_url":"https://api.github.com/users/MechCoder/events{/privacy}","received_events_url":"https://api.github.com/users/MechCoder/received_events","type":"User","site_admin":false},"body":"Python bindings for StreamingKMeans\r\n\r\nWill change status to MRG once docs, tests and examples are updated.","created_at":"2015-05-29T13:32:42Z","updated_at":"2015-06-18T16:53:47Z","closed_at":null,"merged_at":null,"merge_commit_sha":"7884bac84fa6d30a3377646d422a1d6ec3746698","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/6499/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/6499/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/6499/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/81482fd31ff597779724c2ccec7f223aa97a0f60","head":{"label":"MechCoder:spark-4118","ref":"spark-4118","sha":"81482fd31ff597779724c2ccec7f223aa97a0f60","user":{"login":"MechCoder","id":1867024,"avatar_url":"https://avatars.githubusercontent.com/u/1867024?v=3","gravatar_id":"","url":"https://api.github.com/users/MechCoder","html_url":"https://github.com/MechCoder","followers_url":"https://api.github.com/users/MechCoder/followers","following_url":"https://api.github.com/users/MechCoder/following{/other_user}","gists_url":"https://api.github.com/users/MechCoder/gists{/gist_id}","starred_url":"https://api.github.com/users/MechCoder/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/MechCoder/subscriptions","organizations_url":"https://api.github.com/users/MechCoder/orgs","repos_url":"https://api.github.com/users/MechCoder/repos","events_url":"https://api.github.com/users/MechCoder/events{/privacy}","received_events_url":"https://api.github.com/users/MechCoder/received_events","type":"User","site_admin":false},"repo":{"id":23950071,"name":"spark","full_name":"MechCoder/spark","owner":{"login":"MechCoder","id":1867024,"avatar_url":"https://avatars.githubusercontent.com/u/1867024?v=3","gravatar_id":"","url":"https://api.github.com/users/MechCoder","html_url":"https://github.com/MechCoder","followers_url":"https://api.github.com/users/MechCoder/followers","following_url":"https://api.github.com/users/MechCoder/following{/other_user}","gists_url":"https://api.github.com/users/MechCoder/gists{/gist_id}","starred_url":"https://api.github.com/users/MechCoder/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/MechCoder/subscriptions","organizations_url":"https://api.github.com/users/MechCoder/orgs","repos_url":"https://api.github.com/users/MechCoder/repos","events_url":"https://api.github.com/users/MechCoder/events{/privacy}","received_events_url":"https://api.github.com/users/MechCoder/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/MechCoder/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/MechCoder/spark","forks_url":"https://api.github.com/repos/MechCoder/spark/forks","keys_url":"https://api.github.com/repos/MechCoder/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/MechCoder/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/MechCoder/spark/teams","hooks_url":"https://api.github.com/repos/MechCoder/spark/hooks","issue_events_url":"https://api.github.com/repos/MechCoder/spark/issues/events{/number}","events_url":"https://api.github.com/repos/MechCoder/spark/events","assignees_url":"https://api.github.com/repos/MechCoder/spark/assignees{/user}","branches_url":"https://api.github.com/repos/MechCoder/spark/branches{/branch}","tags_url":"https://api.github.com/repos/MechCoder/spark/tags","blobs_url":"https://api.github.com/repos/MechCoder/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/MechCoder/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/MechCoder/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/MechCoder/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/MechCoder/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/MechCoder/spark/languages","stargazers_url":"https://api.github.com/repos/MechCoder/spark/stargazers","contributors_url":"https://api.github.com/repos/MechCoder/spark/contributors","subscribers_url":"https://api.github.com/repos/MechCoder/spark/subscribers","subscription_url":"https://api.github.com/repos/MechCoder/spark/subscription","commits_url":"https://api.github.com/repos/MechCoder/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/MechCoder/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/MechCoder/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/MechCoder/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/MechCoder/spark/contents/{+path}","compare_url":"https://api.github.com/repos/MechCoder/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/MechCoder/spark/merges","archive_url":"https://api.github.com/repos/MechCoder/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/MechCoder/spark/downloads","issues_url":"https://api.github.com/repos/MechCoder/spark/issues{/number}","pulls_url":"https://api.github.com/repos/MechCoder/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/MechCoder/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/MechCoder/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/MechCoder/spark/labels{/name}","releases_url":"https://api.github.com/repos/MechCoder/spark/releases{/id}","created_at":"2014-09-12T06:52:44Z","updated_at":"2015-05-08T08:12:17Z","pushed_at":"2015-06-18T07:29:50Z","git_url":"git://github.com/MechCoder/spark.git","ssh_url":"git@github.com:MechCoder/spark.git","clone_url":"https://github.com/MechCoder/spark.git","svn_url":"https://github.com/MechCoder/spark","homepage":null,"size":104710,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"e86fbdb1e6f1538f65ef78d90bbc41604f6bd580","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-06-18T14:36:17Z","pushed_at":"2015-06-18T16:45:00Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":1487337,"stargazers_count":4429,"watchers_count":4429,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":3894,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":363,"forks":3894,"open_issues":363,"watchers":4429,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/6499"},"html":{"href":"https://github.com/apache/spark/pull/6499"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/6499"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/6499/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/6499/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/6499/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/81482fd31ff597779724c2ccec7f223aa97a0f60"}}}}