[SPARK-4969][STREAMING][PYTHON] Add binaryRecords to streaming #3803

freeman-lab · 2014-12-25T19:46:17Z

In Spark 1.2 we added a binaryRecords input method for loading flat binary data. This format is useful for numerical array data, e.g. in scientific computing applications. This PR adds support for the same format in Streaming applications, where it is similarly useful, especially for streaming time series or sensor data.

Summary of additions

adding binaryRecordsStream to Spark Streaming
exposing binaryRecordsStream in the new PySpark Streaming
new unit tests in Scala and Python

This required adding an optional Hadoop configuration param to fileStream and FileInputStream, but was otherwise straightforward.

@tdas @davies

SparkQA · 2014-12-25T19:47:32Z

Test build #24822 has started for PR 3803 at commit becb344.

This patch merges cleanly.

SparkQA · 2014-12-25T19:48:27Z

Test build #24822 has finished for PR 3803 at commit becb344.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileInputDStream[K, V, F <: NewInputFormat[K,V]](

AmplabJenkins · 2014-12-25T19:48:28Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24822/
Test FAILed.

SparkQA · 2014-12-25T19:57:32Z

Test build #24823 has started for PR 3803 at commit fcb915c.

This patch merges cleanly.

SparkQA · 2014-12-25T21:21:57Z

Test build #24823 has finished for PR 3803 at commit fcb915c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileInputDStream[K, V, F <: NewInputFormat[K,V]](

AmplabJenkins · 2014-12-25T21:22:01Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24823/
Test PASSed.

JoshRosen · 2014-12-26T20:18:25Z

streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala

+      Thread.sleep(1000)
+      // Set up the streaming context and input streams
+      val newConf = conf.clone.set(
+        "spark.streaming.clock", "org.apache.spark.streaming.util.SystemClock")


It looks like this is based on the FileInputStream test, which is known to be flaky. I have a PR open which rewrites that test to not depend on SystemClock / Thread.sleep(): #3801. Therefore, if we want to have this style of test, then this PR should block until my PR is merged so that it can use the new test utilities that I added.

Here's the relevant change from my PR: https://github.com/apache/spark/pull/3801/files#diff-4

Ok great, I'll wait for your PR to be merged and then refactor this test accordingly.

My PR has been merged, so I think this should be unblocked now.

davies · 2015-01-05T18:39:58Z

The Python part look good to me, thanks.

SparkQA · 2015-01-05T22:42:40Z

Test build #25066 has started for PR 3803 at commit 317b6d1.

This patch merges cleanly.

freeman-lab · 2015-01-05T22:56:01Z

Thanks for the review! I'll wait for @JoshRosen 's PR to merge and then update the test here. And will wait for your thoughts on the getBytes issue. Otherwise, I think everything's addressed.

freeman-lab · 2015-02-01T23:41:54Z

Jenkins, retest this please

SparkQA · 2015-02-01T23:42:50Z

Test build #26476 has started for PR 3803 at commit 14bca9a.

This patch merges cleanly.

SparkQA · 2015-02-02T00:53:30Z

Test build #26476 has finished for PR 3803 at commit 14bca9a.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileInputDStream[K, V, F <: NewInputFormat[K,V]](

AmplabJenkins · 2015-02-02T00:53:34Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26476/
Test PASSed.

tdas · 2015-02-02T22:44:57Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

+   * @tparam V Value type for reading HDFS file
+   * @tparam F Input format for reading HDFS file
+   */
+  def fileStream[


Can you make these parameters a superset of the parameters in the other fileStream? Otherwise it seems like people can either use a conf or use a filter

Fine with me, though would require wiring the extra arguments (filter and newFilesOnly) through to all the binary record methods (because they'll also need the conf). Should we expose those arguments in binaryRecords? Or just use their defaults?

Why the wiring to binaryRecords? Those two parameters are only relevant to
DStream, so nothing to pass on down to the RDDs
On Feb 2, 2015 7:30 PM, "Jeremy Freeman" notifications@github.com wrote:

In
streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
#3803 (comment):

@@ -361,6 +363,25 @@ class StreamingContext private[streaming] (

/**
* Create a input stream that monitors a Hadoop-compatible filesystem

* for new files and reads them using the given key-value types and input format.

* Files must be written to the monitored directory by "moving" them from another

* location within the same file system. File names starting with . are ignored.

* @param directory HDFS directory to monitor for new file

* @param conf Hadoop configuration

* @tparam K Key type for reading HDFS file

* @tparam V Value type for reading HDFS file

* @tparam F Input format for reading HDFS file

*/

def fileStream[

Fine with me, though would require wiring the extra arguments (filter and
newFilesOnly) through to all the binary record methods (because they'll
also need the conf). Should we expose those arguments in binaryRecords?
Or just use their defaults?

—
Reply to this email directly or view it on GitHub
https://github.com/apache/spark/pull/3803/files#r23980541.

Sorry, meant exposing them to binaryRecordsStream. That's calling fileStream, and to include conf as an arg we'll also need to either specify filter and newFilesOnly or pass defaults, right?

davies · 2015-02-03T08:29:31Z

The Python parts look good to me, thanks!

- Ensures the argument including conf is a superset of existing arguments - Use default versions for binaryRecordStream

SparkQA · 2015-02-03T22:32:49Z

Test build #26680 has started for PR 3803 at commit eba925c.

This patch merges cleanly.

SparkQA · 2015-02-03T23:44:45Z

Test build #26680 has finished for PR 3803 at commit eba925c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileInputDStream[K, V, F <: NewInputFormat[K,V]](

AmplabJenkins · 2015-02-03T23:44:48Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26680/
Test PASSed.

tdas · 2015-02-04T03:43:08Z

core/src/main/scala/org/apache/spark/SparkContext.scala

+    val data = br.map { case (k, v) =>
+      val bytes = v.getBytes
+      assert(bytes.length == recordLength, "Byte array does not have correct length")
+      bytes


nit: Is this something that the user should be made aware of in the docs?

Do you mean something more than these notes we're adding? I just clarified the notes a bit to make it obvious the check is on the byte array.

I meant should the user be told that the system can throw error when the records are not of the expected size. I dont have any strong feeling on this, just wondering.

Gotcha, I think it's ok as is then. Given what FixedLengthInputFormat is doing, this is more a defensive assertion, it's not something the user should hit due to an inappropriate input.

tdas · 2015-02-04T03:47:07Z

LGTM. Just two minor nits. Can merge as soon as you take a look. This is great @freeman-lab

freeman-lab · 2015-02-04T05:28:00Z

Thanks for the detailed look @tdas! Think I addressed both nits.

SparkQA · 2015-02-04T05:32:55Z

Test build #26725 has started for PR 3803 at commit b676534.

This patch merges cleanly.

SparkQA · 2015-02-04T06:44:59Z

Test build #26725 has finished for PR 3803 at commit b676534.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileInputDStream[K, V, F <: NewInputFormat[K,V]](

AmplabJenkins · 2015-02-04T06:45:03Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26725/
Test PASSed.

tdas · 2015-02-04T06:56:14Z

Thanks Jeremy! Good clean patch!
On Feb 3, 2015 10:45 PM, "UCB AMPLab" notifications@github.com wrote:

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26725/
Test PASSed.

—
Reply to this email directly or view it on GitHub
#3803 (comment).

In Spark 1.2 we added a `binaryRecords` input method for loading flat binary data. This format is useful for numerical array data, e.g. in scientific computing applications. This PR adds support for the same format in Streaming applications, where it is similarly useful, especially for streaming time series or sensor data. Summary of additions - adding `binaryRecordsStream` to Spark Streaming - exposing `binaryRecordsStream` in the new PySpark Streaming - new unit tests in Scala and Python This required adding an optional Hadoop configuration param to `fileStream` and `FileInputStream`, but was otherwise straightforward. tdas davies Author: freeman <the.freeman.lab@gmail.com> Closes #3803 from freeman-lab/streaming-binary-records and squashes the following commits: b676534 [freeman] Clarify note 5ff1b75 [freeman] Add note to java streaming context eba925c [freeman] Simplify notes c4237b8 [freeman] Add experimental tag 30eba67 [freeman] Add filter and newFilesOnly alongside conf c2cfa6d [freeman] Expose new version of fileStream with conf in java 34d20ef [freeman] Add experimental tag 14bca9a [freeman] Add experimental tag b85bffc [freeman] Formatting 47560f4 [freeman] Space formatting 9a3715a [freeman] Refactor to reflect changes to FileInputSuite 7373f73 [freeman] Add note and defensive assertion for byte length 3ceb684 [freeman] Merge remote-tracking branch 'upstream/master' into streaming-binary-records 317b6d1 [freeman] Make test inline fcb915c [freeman] Formatting becb344 [freeman] Formatting d3e75b2 [freeman] Add tests in python a4324a3 [freeman] Line length 029d49c [freeman] Formatting 1c739aa [freeman] Simpler default arg handling 94d90d0 [freeman] Spelling 2843e9d [freeman] Add params to docstring 8b70fbc [freeman] Reorganization 28bff9b [freeman] Fix missing arg 9398bcb [freeman] Expose optional hadoop configuration 23dd69f [freeman] Tests for binaryRecordsStream 36cb0fd [freeman] Add binaryRecordsStream to scala fe4e803 [freeman] Add binaryRecordStream to Java API ecef0eb [freeman] Add binaryRecordsStream to python 8550c26 [freeman] Expose additional argument combination (cherry picked from commit 242b4f0) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

freeman-lab added 15 commits December 25, 2014 04:31

Expose additional argument combination

8550c26

Add binaryRecordsStream to python

ecef0eb

Add binaryRecordStream to Java API

fe4e803

Add binaryRecordsStream to scala

36cb0fd

Tests for binaryRecordsStream

23dd69f

Expose optional hadoop configuration

9398bcb

Fix missing arg

28bff9b

Reorganization

8b70fbc

Add params to docstring

2843e9d

Spelling

94d90d0

Simpler default arg handling

1c739aa

Formatting

029d49c

Line length

a4324a3

Add tests in python

d3e75b2

Formatting

becb344

Formatting

fcb915c

freeman-lab changed the title ~~[SPARK-4969] [STREAMING] [PYTHON] Add binaryRecords to streaming~~ [SPARK-4969][STREAMING][PYTHON] Add binaryRecords to streaming Dec 25, 2014

JoshRosen reviewed Dec 26, 2014
View reviewed changes

Make test inline

317b6d1

tdas reviewed Feb 2, 2015
View reviewed changes

Add experimental tag

34d20ef

freeman-lab added 4 commits February 3, 2015 16:34

Expose new version of fileStream with conf in java

c2cfa6d

Add filter and newFilesOnly alongside conf

30eba67

- Ensures the argument including conf is a superset of existing arguments - Use default versions for binaryRecordStream

Add experimental tag

c4237b8

Simplify notes

eba925c

tdas reviewed Feb 4, 2015
View reviewed changes

freeman-lab added 2 commits February 4, 2015 00:24

Add note to java streaming context

5ff1b75

Clarify note

b676534

asfgit closed this in 242b4f0 Feb 4, 2015

[SPARK-4969][STREAMING][PYTHON] Add binaryRecords to streaming #3803

[SPARK-4969][STREAMING][PYTHON] Add binaryRecords to streaming #3803

Uh oh!

Conversation

freeman-lab commented Dec 25, 2014

Uh oh!

SparkQA commented Dec 25, 2014

Uh oh!

SparkQA commented Dec 25, 2014

Uh oh!

AmplabJenkins commented Dec 25, 2014

Uh oh!

SparkQA commented Dec 25, 2014

Uh oh!

SparkQA commented Dec 25, 2014

Uh oh!

AmplabJenkins commented Dec 25, 2014

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davies commented Jan 5, 2015

Uh oh!

SparkQA commented Jan 5, 2015

Uh oh!

freeman-lab commented Jan 5, 2015

Uh oh!

freeman-lab commented Feb 1, 2015

Uh oh!

SparkQA commented Feb 1, 2015

Uh oh!

SparkQA commented Feb 2, 2015

Uh oh!

AmplabJenkins commented Feb 2, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davies commented Feb 3, 2015

Uh oh!

SparkQA commented Feb 3, 2015

Uh oh!

SparkQA commented Feb 3, 2015

Uh oh!

AmplabJenkins commented Feb 3, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tdas commented Feb 4, 2015

Uh oh!

freeman-lab commented Feb 4, 2015

Uh oh!

SparkQA commented Feb 4, 2015

Uh oh!

SparkQA commented Feb 4, 2015

Uh oh!

AmplabJenkins commented Feb 4, 2015

Uh oh!

tdas commented Feb 4, 2015

Uh oh!

Uh oh!