[SPARK-6328] [Python] Python API for StreamingListener #9186

djalova · 2015-10-20T22:14:41Z

No description provided.

…mplement StreamingListenerEvent

… comment.

tdas · 2015-10-20T23:22:23Z

this is ok to test.

tdas · 2015-10-20T23:22:31Z

@zsxwing can you take a look.

SparkQA · 2015-10-20T23:51:47Z

Test build #44014 has finished for PR 9186 at commit aa87c40.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

djalova · 2015-10-21T00:01:55Z

The test that failed was a flaky StreamingKMeansTest.
Please retest.

tdas · 2015-10-21T01:00:41Z

test this again

djalova · 2015-10-21T18:50:41Z

@tdas it looks like Jenkins didn't retest

zsxwing · 2015-10-22T01:08:25Z

retest this please

SparkQA · 2015-10-22T01:45:10Z

Test build #44115 has finished for PR 9186 at commit aa87c40.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-10-22T06:50:07Z

python/pyspark/streaming/listener.py

+    def onOutputOperationCompleted(self, outputOperationCompleted):
+        pass
+
+    def getEventInfo(self, event):


Is this method necessary?

I thought it would make it easier to see which Info is associated with each StreamingListenerEvent but I can remove it.

Would it be better to use this method to return instances of the Python friendly classes for these Scala objects?

zsxwing · 2015-10-22T06:56:54Z

I suggest adding some Python friendly classes for BatchInfo, ReceiverInfo, OutputOperationInfo and converting these Scala objects to them. Some Scala things, like Scala Map, Option, are hard to use in Python.

… ReceiverInfo

SparkQA · 2015-10-23T19:06:39Z

Test build #44248 has finished for PR 9186 at commit 0ac3df6.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n * class BatchInfo(object):\n * class OutputOperationInfo(object):\n * class ReceiverInfo(object):\n

SparkQA · 2015-10-23T23:42:14Z

Test build #44268 has finished for PR 9186 at commit 233104d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n * class BatchInfo(object):\n * class OutputOperationInfo(object):\n * class ReceiverInfo(object):\n

djalova · 2015-10-24T00:10:18Z

python/pyspark/streaming/tests.py

+        self._test_func(input, func, expected)
+
+        # Test occasionally fails without a delay
+        time.sleep(.1)


The listener sometimes receives 3 batchCompleted events instead of 4 when I run the test, even though it always gets the correct job output. However, when I add a minor delay the test passes consistently. Can I get someone's opinion on this?

Take a look at PySparkStreamingTestCase.wait_for. You can use it to wait for the expected results or timeout.

SparkQA · 2015-10-26T22:07:45Z

Test build #44377 has finished for PR 9186 at commit 79d70ab.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListenerEvent(object):\n * class StreamingListenerBatchSubmitted(StreamingListenerEvent):\n * class StreamingListenerBatchCompleted(StreamingListenerEvent):\n * class StreamingListenerBatchStarted(StreamingListenerEvent):\n * class StreamingListenerOutputOperationStarted(StreamingListenerEvent):\n * class StreamingListenerOutputOperationCompleted(StreamingListenerEvent):\n * class StreamingListenerReceieverStarted(StreamingListenerEvent):\n * class StreamingListenerReceiverError(StreamingListenerEvent):\n * class StreamingListenerReceiverStopped(StreamingListenerEvent):\n * class StreamingListener(object):\n * class StreamingListenerAdapter(StreamingListener):\n * class BatchInfo(object):\n * class OutputOperationInfo(object):\n * class ReceiverInfo(object):\n

zsxwing · 2015-10-27T02:22:50Z

python/pyspark/streaming/listener.py

+    def onBatchStarted(self, batchStarted):
+        batch_info = BatchInfo(batchStarted.batchInfo())
+        batch_started = StreamingListenerBatchStarted(batch_info)
+        self.userStreamingListener .onBatchStarted(batch_started)


nit: redundant space

SparkQA · 2015-11-12T01:17:24Z

Test build #45690 has finished for PR 9186 at commit 7cdaf37.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

zsxwing · 2015-11-12T06:49:33Z

python/pyspark/streaming/tests.py

+            self.assertEqual(info.schedulingDelay(), -1)
+            self.assertEqual(info.processingDelay(), -1)
+            self.assertEqual(info.totalDelay(), -1)
+            self.assertEqual(info.numRecords(), 0)


I just meant you can access the Map here, e.g.:

for streamId in info.streamIdToInputInfo(): streamInputInfo = info.streamIdToInputInfo()[streamId] # access fields of streamInputInfo for outputOpId in info.outputOperationInfos(): outputOperationInfo = info.outputOperationInfos()[outputOpId] # access fields of outputOperationInfo

zsxwing · 2015-11-12T06:52:38Z

I just noticed that Streaming Python unit tests cannot report failure. It always says pass even if some test fails. Investigating it.

zsxwing · 2015-11-12T18:46:54Z

We should retest this PR after merging #9669.

tdas · 2015-11-13T08:31:59Z

#9669 has been merged. please merge with master and test the PR again.

zsxwing · 2015-11-13T17:50:17Z

retest this please

SparkQA · 2015-11-13T18:42:59Z

Test build #45874 has finished for PR 9186 at commit 7cdaf37.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

zsxwing · 2015-11-13T18:53:26Z

@djalova could you fix the test?

[Running <class '__main__.StreamingListenerTests'>]
test_batch_info_reports (__main__.StreamingListenerTests) ... FAIL

======================================================================
FAIL: test_batch_info_reports (__main__.StreamingListenerTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/jenkins/workspace/SparkPullRequestBuilder@2/python/pyspark/streaming/tests.py", line 444, in test_batch_info_reports
    self.assertEqual(len(batchInfosSubmitted), 4)
AssertionError: 5 != 4

----------------------------------------------------------------------
Ran 1 test in 4.768s

FAILED (failures=1)

djalova · 2015-11-13T18:55:51Z

Sure, I'll change it to check for at least 4 batches.

SparkQA · 2015-11-13T20:17:49Z

Test build #45880 has finished for PR 9186 at commit 64192d6.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * public class JavaGradientBoostingClassificationExample\n * public class JavaGradientBoostingRegressionExample\n * public class JavaRandomForestClassificationExample\n * public class JavaRandomForestRegressionExample\n * class StreamingListener(object):\n

zsxwing · 2015-11-13T22:40:52Z

@djalova could you add the following checks?

I just meant you can access the Map here, e.g.:

for streamId in info.streamIdToInputInfo():
streamInputInfo = info.streamIdToInputInfo()[streamId]
# access fields of streamInputInfo

for outputOpId in info.outputOperationInfos():
outputOperationInfo = info.outputOperationInfos()[outputOpId]
# access fields of outputOperationInfo

zsxwing · 2015-11-13T23:13:16Z

Sorry for my unclear comment. I just meant adding the real codes to access fields of streamInputInfo and outputOperationInfo, such as

streamInputInfo.inputStreamId()
streamInputInfo.numRecords()
for key in streamInputInfo.metadata():
    streamInputInfo.metadata()[key]
metadataDescription.metadataDescription()
...

djalova · 2015-11-13T23:40:46Z

Sorry I wasn't reading your comment carefully. I'll make the update.

SparkQA · 2015-11-14T00:01:17Z

Test build #45905 has finished for PR 9186 at commit 5349c82.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

SparkQA · 2015-11-14T01:09:56Z

Test build #45906 has finished for PR 9186 at commit c941c3e.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

zsxwing · 2015-11-14T04:27:10Z

Thanks @djalova LGTM

tdas · 2015-11-16T19:22:05Z

Thanks @djalova and @zsxwing for your efforts. Merging this to master and 1.6

Author: Daniel Jalova <djalova@us.ibm.com> Closes #9186 from djalova/SPARK-6328. (cherry picked from commit ace0db4) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

djalova added 10 commits October 5, 2015 17:12

Added skeleton classes for listener.py

a63ef33

Merge remote-tracking branch 'upstream/master' into SPARK-6328

5b3da81

Added skeleton class for Python API streamingListener

f6c91e9

Added listener.py to __init__.py

1e6a87e

Working implementation of StreamingListener for Python API. Need to i…

2579f64

…mplement StreamingListenerEvent

Modified signature for addStreamingListener to match Scala API, added…

afad086

… comment.

Merge remote-tracking branch 'upstream/master' into SPARK-6328

3523e1f

Refactored StreamingListener methods and added getEventInfo.

7d50848

Added test_batch_info_reports() to streaming/tests.py

9d66e81

Fixed merge conflicts

aa87c40

djalova changed the title ~~[SPARK-6328] [pyspark]~~ [SPARK-6328] [Python] [Streaming] Python API for StreamingListener Oct 20, 2015

djalova changed the title ~~[SPARK-6328] [Python] [Streaming] Python API for StreamingListener~~ [SPARK-6328] [Python] Python API for StreamingListener Oct 20, 2015

zsxwing reviewed Oct 22, 2015
View reviewed changes

Added Python friendly classes for BatchInfo, OutputOperationInfo, and…

0ac3df6

… ReceiverInfo

Fixed pep8 check. Added delay to test_batch_info_reports

233104d

djalova reviewed Oct 24, 2015
View reviewed changes

Added StreamingListenerAdapter for Python.

79d70ab

zsxwing reviewed Oct 27, 2015
View reviewed changes

zsxwing reviewed Nov 12, 2015
View reviewed changes

djalova added 2 commits November 13, 2015 11:02

Fixed asserts in test_batch_info_reports

ddad255

Merge remote-tracking branch 'upstream/master' into SPARK-6328

64192d6

Added more checks to test_batch_info_reports

5349c82

Fixed tests in test_batch_info_reports

c941c3e

asfgit closed this in ace0db4 Nov 16, 2015

djalova deleted the SPARK-6328 branch August 19, 2016 19:23

[SPARK-6328] [Python] Python API for StreamingListener #9186

[SPARK-6328] [Python] Python API for StreamingListener #9186

Conversation

djalova commented Oct 20, 2015

tdas commented Oct 20, 2015

tdas commented Oct 20, 2015

SparkQA commented Oct 20, 2015

djalova commented Oct 21, 2015

tdas commented Oct 21, 2015

djalova commented Oct 21, 2015

zsxwing commented Oct 22, 2015

SparkQA commented Oct 22, 2015

zsxwing Oct 22, 2015

Choose a reason for hiding this comment

djalova Oct 22, 2015

Choose a reason for hiding this comment

djalova Oct 22, 2015

Choose a reason for hiding this comment

zsxwing commented Oct 22, 2015

SparkQA commented Oct 23, 2015

SparkQA commented Oct 23, 2015

djalova Oct 24, 2015

Choose a reason for hiding this comment

zsxwing Oct 27, 2015

Choose a reason for hiding this comment

SparkQA commented Oct 26, 2015

zsxwing Oct 27, 2015

Choose a reason for hiding this comment

SparkQA commented Nov 12, 2015

zsxwing Nov 12, 2015

Choose a reason for hiding this comment

zsxwing commented Nov 12, 2015

zsxwing commented Nov 12, 2015

tdas commented Nov 13, 2015

zsxwing commented Nov 13, 2015

SparkQA commented Nov 13, 2015

zsxwing commented Nov 13, 2015

djalova commented Nov 13, 2015

SparkQA commented Nov 13, 2015

zsxwing commented Nov 13, 2015

zsxwing commented Nov 13, 2015

djalova commented Nov 13, 2015

SparkQA commented Nov 14, 2015

SparkQA commented Nov 14, 2015

zsxwing commented Nov 14, 2015

tdas commented Nov 16, 2015