You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by kanzhang <gi...@git.apache.org> on 2014/07/09 07:30:57 UTC
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
GitHub user kanzhang opened a pull request:
https://github.com/apache/spark/pull/1338
[SPARK-2024] Add saveAsSequenceFile and saveAsHadoopFile to PySpark
JIRA issue: https://issues.apache.org/jira/browse/SPARK-2024
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kanzhang/spark SPARK-2024
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1338.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1338
----
commit 8fe613949d2a0b82666567767405eadea9c1c7f1
Author: Kan Zhang <kz...@apache.org>
Date: 2014-07-08T05:37:56Z
[SPARK-2024] Add saveAsSequenceFile to PySpark
commit eef0b2bdd3e704c010744858d54d69adfd8056eb
Author: Kan Zhang <kz...@apache.org>
Date: 2014-07-08T19:33:13Z
[SPARK-2024] More refactoring and documentation
commit e017709cb526d2899676a9ab24c9891da1a20f59
Author: Kan Zhang <kz...@apache.org>
Date: 2014-07-09T03:09:01Z
[SPARK-2024] More Python API documentation
commit 8083ea529dfb374a00c8fe9ae5b7e32189f8147e
Author: Kan Zhang <kz...@apache.org>
Date: 2014-07-09T05:22:16Z
[SPARK-2024] Add programming guide
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50580086
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17424/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r14854296
--- Diff: docs/programming-guide.md ---
@@ -403,31 +403,30 @@ PySpark SequenceFile support loads an RDD within Java, and pickles the resulting
<tr><td>BooleanWritable</td><td>bool</td></tr>
<tr><td>BytesWritable</td><td>bytearray</td></tr>
<tr><td>NullWritable</td><td>None</td></tr>
-<tr><td>ArrayWritable</td><td>list of primitives, or tuple of objects</td></tr>
--- End diff --
Did this work before and get removed now, or was it a mistake in the docs?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48818519
This looks awesome, thanks for putting it together! One comment I have though is that we should add more test coverage, to make sure we cover all the data types supported. Instead of doing this in doc comments, which gets unwieldy, you can do it in python/pyspark/tests.py, which is a standalone test file. Just make sure we have tests that cover each supported data type in sequence files.
@MLnick you should look at this too when you have a chance.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48508590
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16468/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15601428
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -65,20 +66,49 @@ private[python] object SerDeUtil extends Logging {
* by PySpark. By default, if serialization fails, toString is called and the string
* representation is serialized
*/
- def rddToPython(rdd: RDD[(Any, Any)]): RDD[Array[Byte]] = {
+ def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
val (keyFailed, valueFailed) = checkPickle(rdd.first())
rdd.mapPartitions { iter =>
val pickle = new Pickler
- iter.map { case (k, v) =>
- if (keyFailed && valueFailed) {
- pickle.dumps(Array(k.toString, v.toString))
- } else if (keyFailed) {
- pickle.dumps(Array(k.toString, v))
- } else if (!keyFailed && valueFailed) {
- pickle.dumps(Array(k, v.toString))
- } else {
- pickle.dumps(Array(k, v))
+ val cleaned = iter.map { case (k, v) =>
+ val key = if (keyFailed) k.toString else k
+ val value = if (valueFailed) v.toString else v
+ Array[Any](key, value)
+ }
+ if (batchSize > 1) {
+ cleaned.grouped(batchSize).map(batched => pickle.dumps(seqAsJavaList(batched)))
+ } else {
+ cleaned.map(pickle.dumps(_))
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple (K, V) to RDD[(K, V)].
+ */
+ def pythonToPairRDD[K, V](pyRDD: RDD[Array[Byte]]): RDD[(K, V)] = {
+ def isPair(obj: Any): Boolean = {
+ Option(obj.getClass.getComponentType).map(!_.isPrimitive).getOrElse(false) &&
+ obj.asInstanceOf[Array[_]].length == 2
+ }
+ pyRDD.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // batch serialized Python RDDs
+ case objs: java.util.List[_] => objs
--- End diff --
This also affects `PythonRDD.pythonToJavaMap(); good catch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50093703
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17141/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49908850
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17042/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50313472
This looks good to me - I'm +1 to merge.
@JoshRosen @mateiz any further comments?
Thanks @kanzhang for all the work!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50547384
Nevermind. I'm refactoring.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15296619
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -31,13 +31,14 @@ import org.apache.spark.annotation.Experimental
* transformation code by overriding the convert method.
*/
@Experimental
-trait Converter[T, U] extends Serializable {
+trait Converter[T, +U] extends Serializable {
--- End diff --
@mateiz thanks, Matei. I saw it but I couldn't believe that was the reason :-). I added the + sign because some of our converters have more specific types like [Any, Writable] and the compiler complains when assigning them to where [Any, Any] is required. I don't have a strong preference here and could change them back to [Any, Any]. Let me know.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50197882
The latest update fixed JevaBean test failure. To use ```WritableUtils.clone```, I broadcasted Hadoop conf the second time. Is there a way to share the one used by hadoopRDD?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49912846
The latest test failure in ```org.apache.spark.streaming.CheckpointSuite``` seems unrelated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50574875
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17419/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50545354
Now I have got the following error, since ```saveAsHadoopFile``` has 11 params. Can relax it a bit?
```
Scalastyle checks failed at following occurrences:
error file=/home/jenkins/workspace/SparkPullRequestBuilder@3/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala message=The number of parameters should not exceed 10 line=627 column=6
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50386062
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17292/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50090301
Update addressing serialization issue raised by Nick, along with tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50668773
> I think we should remove the batchSerialized arguments from PythonRDD's saveAs* methods and add a batchSerialized field to PythonRDD's constructor, since it's an attribute of the RDD itself rather than an option.
Problem with that is currently PythonRDD objects are only created by PipelinedRDD, whereas in other cases (e.g., ```PythonRDD.readRDDFromFile``` and ```SchemaRDD.javaToPython```), ```_jrdd``` (or ```JavaRDD[Array[Byte]]```) are created directly without PythonRDD objects. I feel the change to use PythonRDD everywhere is too big for this patch. Maybe a followup JIRA?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50378599
For completeness, added one ```saveAsHadoopDataset()``` and one ```saveAsNewAPIHadoopDataset()``` test.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49807264
@MLnick I merely removed it from programming guide. The functionality (and your test) is still there should anyone wants to try it.
@mateiz I meant when reading JavaBeans, you get a dict of attributes to values on the Python side. But you can't turn around and save it as JavaBeans from Python. What you save is a Java Map since that's what Pyrolite will pickle a dict to. I was trying to confirm the same asymmetry on the saving side (i.e., saving a Python custom object as Java Map, and reads it back as a Python dict), but I got the following exception and gave up.
```
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.saveAsSequenceFile.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 33.0:2 failed 1 times, most recent failure: Exception failure in TID 70 on host localhost: net.razorvine.pickle.InvalidOpcodeException: opcode not implemented: OBJ
net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:223)
net.razorvine.pickle.Unpickler.load(Unpickler.java:84)
net.razorvine.pickle.Unpickler.loads(Unpickler.java:97)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50379059
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17292/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48818358
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16588/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50203116
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17195/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50208947
@kanzhang I think it's unsafe to unpersist the read job's Hadoop configuration. Even if we cached the results of the read, I think we'd still need to keep a copy of the configuration in case any of those cached partitions are lost and we need to recompute from scratch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50547420
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17386/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49808256
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16994/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15508664
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -65,20 +66,49 @@ private[python] object SerDeUtil extends Logging {
* by PySpark. By default, if serialization fails, toString is called and the string
* representation is serialized
*/
- def rddToPython(rdd: RDD[(Any, Any)]): RDD[Array[Byte]] = {
+ def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
val (keyFailed, valueFailed) = checkPickle(rdd.first())
rdd.mapPartitions { iter =>
val pickle = new Pickler
- iter.map { case (k, v) =>
- if (keyFailed && valueFailed) {
- pickle.dumps(Array(k.toString, v.toString))
- } else if (keyFailed) {
- pickle.dumps(Array(k.toString, v))
- } else if (!keyFailed && valueFailed) {
- pickle.dumps(Array(k, v.toString))
- } else {
- pickle.dumps(Array(k, v))
+ val cleaned = iter.map { case (k, v) =>
+ val key = if (keyFailed) k.toString else k
+ val value = if (valueFailed) v.toString else v
+ Array[Any](key, value)
+ }
+ if (batchSize > 1) {
+ cleaned.grouped(batchSize).map(batched => pickle.dumps(seqAsJavaList(batched)))
+ } else {
+ cleaned.map(pickle.dumps(_))
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple (K, V) to RDD[(K, V)].
+ */
+ def pythonToPairRDD[K, V](pyRDD: RDD[Array[Byte]]): RDD[(K, V)] = {
+ def isPair(obj: Any): Boolean = {
+ Option(obj.getClass.getComponentType).map(!_.isPrimitive).getOrElse(false) &&
+ obj.asInstanceOf[Array[_]].length == 2
+ }
+ pyRDD.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // batch serialized Python RDDs
+ case objs: java.util.List[_] => objs
--- End diff --
What if I disabled batching and serialized an RDD of Python lists? In that case, I think this would end up inadvertently flattening the RDD of lists. To handle this case, I think you need to store an additional field in PythonRDD to record whether its pickled objects were batch serialized.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50220618
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17213/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50225209
QA results for PR 1338:<br>- This patch PASSES unit tests.<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17216/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15555021
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -65,20 +66,49 @@ private[python] object SerDeUtil extends Logging {
* by PySpark. By default, if serialization fails, toString is called and the string
* representation is serialized
*/
- def rddToPython(rdd: RDD[(Any, Any)]): RDD[Array[Byte]] = {
+ def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
val (keyFailed, valueFailed) = checkPickle(rdd.first())
rdd.mapPartitions { iter =>
val pickle = new Pickler
- iter.map { case (k, v) =>
- if (keyFailed && valueFailed) {
- pickle.dumps(Array(k.toString, v.toString))
- } else if (keyFailed) {
- pickle.dumps(Array(k.toString, v))
- } else if (!keyFailed && valueFailed) {
- pickle.dumps(Array(k, v.toString))
- } else {
- pickle.dumps(Array(k, v))
+ val cleaned = iter.map { case (k, v) =>
+ val key = if (keyFailed) k.toString else k
+ val value = if (valueFailed) v.toString else v
+ Array[Any](key, value)
+ }
+ if (batchSize > 1) {
+ cleaned.grouped(batchSize).map(batched => pickle.dumps(seqAsJavaList(batched)))
+ } else {
+ cleaned.map(pickle.dumps(_))
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple (K, V) to RDD[(K, V)].
+ */
+ def pythonToPairRDD[K, V](pyRDD: RDD[Array[Byte]]): RDD[(K, V)] = {
+ def isPair(obj: Any): Boolean = {
+ Option(obj.getClass.getComponentType).map(!_.isPrimitive).getOrElse(false) &&
+ obj.asInstanceOf[Array[_]].length == 2
+ }
+ pyRDD.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ iter.flatMap { row =>
+ unpickle.loads(row) match {
+ // batch serialized Python RDDs
+ case objs: java.util.List[_] => objs
--- End diff --
You are absolutely right. I fixed it by passing in a flag from the caller, since PythonRDD is not always created (e.g., in ```sc.parallelize()```). It also helped me finding a bug in ```_toPickleSerialization()```. Let me know if you have any further comments. Thx.
This issue may also affect ```PythonRDD.pythonToJavaMap()``` used by SchemaRDD. @davies , it might be relevant to your namedtuple work.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48971710
@MLnick I'll see if I can add couple output converter examples as well. Thx.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50210598
> Even if we cached the results of the read, I think we'd still need to keep a copy of the configuration in case any of those cached partitions are lost and we need to recompute from scratch.
This is where I have a gap. I imagined the error recovery code would have kept a copy of the conf on driver and re-broadcast it when re-computation happens, whereas those broadcasted copies on executors could have been cleaned after job is successfully run the first time.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50186945
The batch size is a performance-tuning knob that controls the granularity of data transfer between Python and Java. Larger batch sizes help to amortize certain costs associating with pickling / depickling and in storing pickled objects in byte arrays. If the batch size is too large, the Python worker may run out of memory while trying to serialize or deserialize too much data at once.
A batch size of _n_ means that we emit a new batch every _n_ items, but this doesn't guarantee that every batch will contain exactly _n_ items (for example, if the last (or only) batch might contain fewer than _n_ items). As a result, the batch deserializer doesn't make any assumptions about how many items appear in a batch, so it's safe to union two batched datasets as long as they have the same underlying serializer (such as PickleSerializer).
I don't remember why we chose these default batch sizes; I suppose we could run some experiments to sweep across different batch sizes, but I imagine that the optimal size depends on the data type.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15607176
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -65,20 +66,52 @@ private[python] object SerDeUtil extends Logging {
* by PySpark. By default, if serialization fails, toString is called and the string
* representation is serialized
*/
- def rddToPython(rdd: RDD[(Any, Any)]): RDD[Array[Byte]] = {
+ def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
val (keyFailed, valueFailed) = checkPickle(rdd.first())
rdd.mapPartitions { iter =>
val pickle = new Pickler
- iter.map { case (k, v) =>
- if (keyFailed && valueFailed) {
- pickle.dumps(Array(k.toString, v.toString))
- } else if (keyFailed) {
- pickle.dumps(Array(k.toString, v))
- } else if (!keyFailed && valueFailed) {
- pickle.dumps(Array(k, v.toString))
+ val cleaned = iter.map { case (k, v) =>
+ val key = if (keyFailed) k.toString else k
+ val value = if (valueFailed) v.toString else v
+ Array[Any](key, value)
+ }
+ if (batchSize > 1) {
+ cleaned.grouped(batchSize).map(batched => pickle.dumps(seqAsJavaList(batched)))
+ } else {
+ cleaned.map(pickle.dumps(_))
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple (K, V) to RDD[(K, V)].
+ */
+ def pythonToPairRDD[K, V](pyRDD: RDD[Array[Byte]], batchSerialized: Boolean): RDD[(K, V)] = {
+ def isPair(obj: Any): Boolean = {
+ Option(obj.getClass.getComponentType).map(!_.isPrimitive).getOrElse(false) &&
+ obj.asInstanceOf[Array[_]].length == 2
+ }
+ pyRDD.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ val unpickled =
--- End diff --
Can we defer this refactoring to when we update ```pythonToJavaMap```, since I don't want to touch SchemaRDD code in this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49895962
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17042/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50551599
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17389/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50090439
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17141/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15386068
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -92,6 +104,46 @@ private[python] class DefaultConverter extends Converter[Any, Any] {
}
}
+/**
+ * A converter that converts common types to [[org.apache.hadoop.io.Writable]]. Note that array
+ * types are not supported since the user needs to subclass [[org.apache.hadoop.io.ArrayWritable]]
+ * to set the type properly. See [[org.apache.spark.api.python.DoubleArrayWritable]] and
+ * [[org.apache.spark.api.python.DoubleArrayToWritableConverter]] for an example. They are used in
+ * PySpark RDD `saveAsNewAPIHadoopFile` doctest.
+ */
+private[python] class JavaToWritableConverter extends Converter[Any, Writable] {
+
+ /**
+ * Converts common data types to [[org.apache.hadoop.io.Writable]]. Note that array types are not
+ * supported out-of-the-box.
+ */
+ private def convertToWritable(obj: Any): Writable = {
+ import collection.JavaConversions._
+ obj match {
+ case i: java.lang.Integer => new IntWritable(i)
+ case d: java.lang.Double => new DoubleWritable(d)
+ case l: java.lang.Long => new LongWritable(l)
+ case f: java.lang.Float => new FloatWritable(f)
+ case s: java.lang.String => new Text(s)
+ case b: java.lang.Boolean => new BooleanWritable(b)
+ case aob: Array[Byte] => new BytesWritable(aob)
+ case null => NullWritable.get()
+ case map: java.util.Map[_, _] =>
+ val mapWritable = new MapWritable()
+ map.foreach { case (k, v) =>
+ mapWritable.put(convertToWritable(k), convertToWritable(v))
+ }
+ mapWritable
+ case other => throw new SparkException(s"Data of type $other cannot be used")
--- End diff --
This comment also applies to the other unsupported type messages added in this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50544017
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17384/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50111889
@kanzhang it might mean that we're reusing the Bean object on the Java side when we read from the InputFormat. Hadoop's RecordReaders actually reuse the same object as you read data, so if you want to hold onto multiple data items, you need to clone each one. This may be a fair bit of trouble unfortunately.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r14854311
--- Diff: python/pyspark/tests.py ---
@@ -273,6 +274,14 @@ def test_sequencefiles(self):
(3, {2.0: u'dd'})]
self.assertEqual(maps, em)
+ arrays = sorted(self.sc.sequenceFile(basepath + "/sftestdata/sfarray/",
+ "org.apache.hadoop.io.IntWritable",
+ "org.apache.spark.api.python.DoubleArrayWritable").collect())
--- End diff --
This line looks longer than 100 characters, change the indentation a bit to prevent that
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50202364
Just wanted to make sure my understanding is correct, calling ```RDD.first()``` doesn't materialize (read in) the full RDD, right? I saw a ```first``` being called within ```SerDeUtil.pairRDDToPython``` and was wondering if I could ```unpersist``` the broadcasted Hadoop conf after that and before returning from ```PythonRDD.sequenceFile```?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48512829
@pwendell the 2 test failures seem unrelated?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15387719
--- Diff: python/pyspark/rdd.py ---
@@ -964,6 +964,106 @@ def first(self):
"""
return self.take(1)[0]
+ def saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None):
+ """
+ Output a Python RDD of key-value pairs (of form C{RDD[(K, V)]}) to any Hadoop file
+ system, using the new Hadoop OutputFormat API (mapreduce package). Keys/values are
+ converted for output using either user specified converters or, by default,
+ L{org.apache.spark.api.python.JavaToWritableConverter}.
+
+ @param conf: Hadoop job configuration, passed in as a dict
+ @param keyConverter: (None by default)
+ @param valueConverter: (None by default)
+ """
+ jconf = self.ctx._dictToJavaMap(conf)
+ reserialized = self._reserialize(BatchedSerializer(PickleSerializer(), 10))
--- End diff --
Since this 10 appears in a lot of places, maybe factor it out into a constant?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49554126
@MLnick I'm thinking of removing the tests and programming guide entry for custom classes (JavaBeans). It seems to be a feature of Pyrolite and I can't think of any obvious use of it in the context of RDDs. For example, Pyrolite maps a JavaBean to a dict of its attributes in Python, but one can't go reverse. Listing it as a supported data type may add confusion to users. Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15388452
--- Diff: python/pyspark/rdd.py ---
@@ -964,6 +964,106 @@ def first(self):
"""
return self.take(1)[0]
+ def saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None):
+ """
+ Output a Python RDD of key-value pairs (of form C{RDD[(K, V)]}) to any Hadoop file
+ system, using the new Hadoop OutputFormat API (mapreduce package). Keys/values are
+ converted for output using either user specified converters or, by default,
+ L{org.apache.spark.api.python.JavaToWritableConverter}.
+
+ @param conf: Hadoop job configuration, passed in as a dict
+ @param keyConverter: (None by default)
+ @param valueConverter: (None by default)
+ """
+ jconf = self.ctx._dictToJavaMap(conf)
+ reserialized = self._reserialize(BatchedSerializer(PickleSerializer(), 10))
--- End diff --
@kanzhang @JoshRosen Is 10 a reasonable default? If we're adding this to all the reading and writing methods, should we not make it an argument rather with default 10?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48501142
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50198112
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17195/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15385187
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -92,6 +104,46 @@ private[python] class DefaultConverter extends Converter[Any, Any] {
}
}
+/**
+ * A converter that converts common types to [[org.apache.hadoop.io.Writable]]. Note that array
+ * types are not supported since the user needs to subclass [[org.apache.hadoop.io.ArrayWritable]]
+ * to set the type properly. See [[org.apache.spark.api.python.DoubleArrayWritable]] and
+ * [[org.apache.spark.api.python.DoubleArrayToWritableConverter]] for an example. They are used in
+ * PySpark RDD `saveAsNewAPIHadoopFile` doctest.
+ */
+private[python] class JavaToWritableConverter extends Converter[Any, Writable] {
+
+ /**
+ * Converts common data types to [[org.apache.hadoop.io.Writable]]. Note that array types are not
+ * supported out-of-the-box.
+ */
+ private def convertToWritable(obj: Any): Writable = {
+ import collection.JavaConversions._
+ obj match {
+ case i: java.lang.Integer => new IntWritable(i)
+ case d: java.lang.Double => new DoubleWritable(d)
+ case l: java.lang.Long => new LongWritable(l)
+ case f: java.lang.Float => new FloatWritable(f)
+ case s: java.lang.String => new Text(s)
+ case b: java.lang.Boolean => new BooleanWritable(b)
+ case aob: Array[Byte] => new BytesWritable(aob)
+ case null => NullWritable.get()
+ case map: java.util.Map[_, _] =>
+ val mapWritable = new MapWritable()
+ map.foreach { case (k, v) =>
+ mapWritable.put(convertToWritable(k), convertToWritable(v))
+ }
+ mapWritable
+ case other => throw new SparkException(s"Data of type $other cannot be used")
--- End diff --
It's probably better to log `${other.getClass.getName}` instead of `$other`, since the string representation of the object may not identify its class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50225107
Rebased just now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50238113
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17224/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by ash211 <gi...@git.apache.org>.
Github user ash211 commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50112894
Matei, what InputFormats did you have problems with when cloning by
default? I'd love to figure out what it would take to solve the one
element/one value problem.
On Thu, Jul 24, 2014 at 11:26 PM, Matei Zaharia <no...@github.com>
wrote:
> BTW you can try WritableUtils.clone. At some point we tried cloning data
> by default in hadoopRDD, or having a flag for it, and we gave up because it
> didn't seem to work for every InputFormat. But it's probably worth a shot
> here if the object extends Writable.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/1338#issuecomment-50112244>.
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50577245
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17419/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15386313
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -92,6 +104,46 @@ private[python] class DefaultConverter extends Converter[Any, Any] {
}
}
+/**
+ * A converter that converts common types to [[org.apache.hadoop.io.Writable]]. Note that array
+ * types are not supported since the user needs to subclass [[org.apache.hadoop.io.ArrayWritable]]
+ * to set the type properly. See [[org.apache.spark.api.python.DoubleArrayWritable]] and
+ * [[org.apache.spark.api.python.DoubleArrayToWritableConverter]] for an example. They are used in
+ * PySpark RDD `saveAsNewAPIHadoopFile` doctest.
+ */
+private[python] class JavaToWritableConverter extends Converter[Any, Writable] {
+
+ /**
+ * Converts common data types to [[org.apache.hadoop.io.Writable]]. Note that array types are not
+ * supported out-of-the-box.
+ */
+ private def convertToWritable(obj: Any): Writable = {
+ import collection.JavaConversions._
+ obj match {
+ case i: java.lang.Integer => new IntWritable(i)
+ case d: java.lang.Double => new DoubleWritable(d)
+ case l: java.lang.Long => new LongWritable(l)
+ case f: java.lang.Float => new FloatWritable(f)
+ case s: java.lang.String => new Text(s)
+ case b: java.lang.Boolean => new BooleanWritable(b)
+ case aob: Array[Byte] => new BytesWritable(aob)
+ case null => NullWritable.get()
+ case map: java.util.Map[_, _] =>
+ val mapWritable = new MapWritable()
+ map.foreach { case (k, v) =>
+ mapWritable.put(convertToWritable(k), convertToWritable(v))
+ }
+ mapWritable
+ case other => throw new SparkException(s"Data of type $other cannot be used")
--- End diff --
@JoshRosen thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50068945
Ok I've been through it in detail and everything looks good, with the exception of one issue that I picked up. The saving currently inherently assumes that the `serializer` on the `PythonRDD` is `PickleSerializer`. If this is not the case for whatever reason then it will fail.
This happens for `RDD`s created from `cartesian` and `zip`:
```python
In [1]: rdd = sc.parallelize(range(10))
In [2]: cart = rdd.cartesian(rdd)
In [3]: cart.saveAsSequenceFile("/tmp/cart.sf")
---------------------------------------------------------------------------
Py4JError Traceback (most recent call last)
<ipython-input-3-06070310df19> in <module>()
----> 1 cart.saveAsSequenceFile("/tmp/cart.sf")
...
Py4JError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.saveAsSequenceFile. Trace:
py4j.Py4JException: Method saveAsSequenceFile([class org.apache.spark.api.java.JavaPairRDD, class java.lang.String, null]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:333)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:358)
at py4j.Gateway.invoke(Gateway.java:254)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
In [4]: x = sc.parallelize(range(0,5))
In [5]: y = sc.parallelize(range(1000, 1005))
In [6]: x.zip(y).saveAsSequenceFile("/tmp/zip.sf")
---------------------------------------------------------------------------
Py4JError Traceback (most recent call last)
<ipython-input-6-ad82cb2565a3> in <module>()
----> 1 x.zip(y).saveAsSequenceFile("/tmp/zip.sf")
...
/Users/Nick/workspace/scala/spark/python/lib/py4j-0.8.1-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
302 raise Py4JError(
303 'An error occurred while calling {0}{1}{2}. Trace:\n{3}\n'.
--> 304 format(target_id, '.', name, value))
305 else:
306 raise Py4JError(
Py4JError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.saveAsSequenceFile. Trace:
py4j.Py4JException: Method saveAsSequenceFile([class org.apache.spark.api.java.JavaPairRDD, class java.lang.String, null]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:333)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:358)
at py4j.Gateway.invoke(Gateway.java:254)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:745)
```
The fix is simple - use what `saveAsPickleFile` does:
```python
self._reserialize(BatchedSerializer(PickleSerializer(), batchSize))._jrdd
```
I confirmed that something like the following works in `saveAsSequenceFile`:
```python
reser_rdd = self._reserialize(BatchedSerializer(PickleSerializer(), batchSize))._jrdd
self.ctx._jvm.PythonRDD.saveAsSequenceFile(reser_rdd, path, compressionCodecClass)
```
Apart from this I think this looks good for merge, subject to @mateiz call on the covariance point.
@kanzhang would be great to try get this finalised for merge window tomorrow. Sorry for the delay on my side!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50211345
Whoops, I was thinking of `destroy` instead of `unpersist.` Since the driver keeps a copy of the broadcast variable, it should always be safe to unpersist.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1338
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50672937
Ah, I see. I don't mind deferring that refactoring to a later patch. I'll create some PySpark refactoring JIRAs later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49843828
The style check error is different, see https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16994/consoleFull. It's a bit hidden in there but it says:
```
error file=/home/jenkins/workspace/SparkPullRequestBuilder@7/core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala message=There should be a space after the plus (+) sign line=34 column=19
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48501125
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50208305
@JoshRosen BTW, thanks for the detailed explanation on serialization!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48820160
I have had a quick look over and will try to do a more detailed one this weekend.
High level looks good, 2 comments so far:
1. Agree with Matei that I think the tests should live in `tests.py` as opposed to docstrings, and add tests for other datatypes in a similar manner to the input format tests
2. Would be great to add a couple of examples of using the custom `Converter` in reverse for output. Again, a Cassandra and HBase example in similar vein to the input format examples would be valuable I think
Will provide any more feedback as I go through it in more detail.
(btw thanks for fixing up the `ArrayWritable` stuff too).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50544121
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17384/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15150914
--- Diff: docs/programming-guide.md ---
@@ -403,31 +403,30 @@ PySpark SequenceFile support loads an RDD within Java, and pickles the resulting
<tr><td>BooleanWritable</td><td>bool</td></tr>
<tr><td>BytesWritable</td><td>bytearray</td></tr>
<tr><td>NullWritable</td><td>None</td></tr>
-<tr><td>ArrayWritable</td><td>list of primitives, or tuple of objects</td></tr>
--- End diff --
@mateiz we don't handle arrays currently and this is also the case for Scala API. The reason is ArrayWritable class doesn't have a no-arg constructor for creating an empty instance upon reading. User needs to create subtypes. Although we could add subtypes for handling primitive arrays, that makes Spark a dependency for users, which we probably don't want to do.
For conversion between arrays and ArrayWritable subtypes, we can deserialize automatically upon reading as long as the subtype is on the class path. However, our default converter can't convert arrays to ArrayWritable subtypes upon writing since we don't know which subtype to use. User needs to specify custom converters.
We should look into ArrayPrimitiveWritable, which is not available in Hadoop v1.0.4.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50110269
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17164/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49708463
@kanzhang @mateiz Yeah this is one issue with Pyrolite vs MsgPack. MsgPack supported case classes out the box, which would likely be a bit more common that beans.
I'd say that custom serde via `Converter` will be far more common (as we've already seen with various Avro commentary etc).
Thinking about it some more, I would be ok to remove from the docs. This would still be available as undocumented functionality so if relevant use cases did come up on the mailing list, we could point to it and in the unlikely case that there was demand we could simply document it as read-only functionality.
Bearing in mind this is also still marked experimental and we'll need to see how users use it in the wild a bit and make any amendments as required.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49810291
Major changes for the updated patch.
1. Replaced doctests with standalone tests
2. Fixed converter for converting BytesWritables and added read/write tests for BytesWritable and byte arrays
3. Added HBase and Cassandra output format and converter examples
4. I used to inspect array element types and try to convert Object[] to array of primitive types whenever possible (so that they get pickled to Python arrays, whereas Object[] gets pickled to Python tuples). But I removed that code, since I can't determine element types for empty arrays. Users have to supply custom converters if they want Java arrays to appear as Python arrays (and they can do that since they know their array types a priori).
5. No out-of-box support for reading/writing arrays, since ArrayWritable itself doesn't have a no-arg constructor for creating an empty instance upon deserializing. Users need to provide ArrayWritable subtypes. Custom converters for converting arrays to suitable ArrayWritable subtypes are also needed when writing. When reading, the default converter will convert any custom ArrayWritable subtypes to Object[] and they get pickled to Python tuples.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50240323
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17224/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48433560
@MLnick @mateiz @ahirreddy @pwendell
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15153275
--- Diff: docs/programming-guide.md ---
@@ -403,31 +403,30 @@ PySpark SequenceFile support loads an RDD within Java, and pickles the resulting
<tr><td>BooleanWritable</td><td>bool</td></tr>
<tr><td>BytesWritable</td><td>bytearray</td></tr>
<tr><td>NullWritable</td><td>None</td></tr>
-<tr><td>ArrayWritable</td><td>list of primitives, or tuple of objects</td></tr>
--- End diff --
We can't write arrays in Scala either (the implicit conversion from Array to ArrayWritable is marked private). Otherwise, it can be awkward as we can't read it back since ArrayWritable doesn't have a no-arg constructor. For user supplied ArrayWritable subtypes, we can read them, it's just they won't be implicitly converted. Essentially the same support as we have in Python.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15275379
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -31,13 +31,14 @@ import org.apache.spark.annotation.Experimental
* transformation code by overriding the convert method.
*/
@Experimental
-trait Converter[T, U] extends Serializable {
+trait Converter[T, +U] extends Serializable {
--- End diff --
Actually the style checker seems to be complaining about this +, which is a mistake in the style checker. You can add a space after the + for now. But do we really need covariance here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50108508
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17164/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49808354
QA results for PR 1338:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16994/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49675615
Regarding the JavaBeans, is there a reason to believe Pyrolite won't support them in the future? Or are you just suggesting to remove it because we can't also save data? That would be a bit of a regression for the reading side, though maybe InputFormats that return JavaBeans are not that common.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48508588
Merged build finished.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50116221
@ash211 this was the JIRA: https://issues.apache.org/jira/browse/SPARK-1018. I don't remember the problematic types exactly, but one might've been Avro records, so try that (I remember there were a bunch of issues with them).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50112244
BTW you can try WritableUtils.clone. At some point we tried cloning data by default in hadoopRDD, or having a flag for it, and we gave up because it didn't seem to work for every InputFormat. But it's probably worth a shot here if the object extends Writable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48433563
Merged build finished.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48818290
Jenkins, retest this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48820892
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br> <td>dict of public properties (via JavaBean getters and setters) __class__ for the class type</td></tr><br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16588/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50555479
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17389/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48433564
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16446/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50093417
There is one additional issue though. Currently, when reading the Python rdd is not batch serialized. This will trigger re-serialization upon writing the same rdd, which should be avoided. Will upload a fix.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50225130
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17218/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15388927
--- Diff: python/pyspark/rdd.py ---
@@ -964,6 +964,106 @@ def first(self):
"""
return self.take(1)[0]
+ def saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None):
+ """
+ Output a Python RDD of key-value pairs (of form C{RDD[(K, V)]}) to any Hadoop file
+ system, using the new Hadoop OutputFormat API (mapreduce package). Keys/values are
+ converted for output using either user specified converters or, by default,
+ L{org.apache.spark.api.python.JavaToWritableConverter}.
+
+ @param conf: Hadoop job configuration, passed in as a dict
+ @param keyConverter: (None by default)
+ @param valueConverter: (None by default)
+ """
+ jconf = self.ctx._dictToJavaMap(conf)
+ reserialized = self._reserialize(BatchedSerializer(PickleSerializer(), 10))
--- End diff --
Making it configurable would also give users a way to avoid the clone issue with types that WritableUtils.clone doesn't work on, so I like that. They'd have to configure it to 1 and we'd have to make sure we don't try to clone in that case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48783095
LGTM, awesome!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50225958
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17218/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50442216
Looks good to me as well. @JoshRosen any comments?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50219459
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17213/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50078131
@MLnick Thanks, Nick! Oops, I totally forgot about re-serialization, which I specifically dealt with when I worked on ```saveAsPickleFile```. Will update now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50382278
Refactoring newly added tests to use corresponding hadoopRDD API to read.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50656448
I think we should remove the `batchSerialized` arguments from PythonRDD's `saveAs*` methods and add a `batchSerialized` field to PythonRDD's constructor, since it's an attribute of the RDD itself rather than an option.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50111934
The same thing happens in normal Spark if you create a hadoopRDD or sequenceFile with Writables inside it, and then call cache(). There will be only one key element and one value, so all the data will look identical.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48962322
@davies @mateiz @MLnick thanks for the review and suggestions. I'll try to add standalone tests for every data type.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48431564
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50183228
> it might mean that we're reusing the Bean object on the Java side when we read from the InputFormat. Hadoop's RecordReaders actually reuse the same object as you read data, so if you want to hold onto multiple data items, you need to clone each one. This may be a fair bit of trouble unfortunately.
That should explain it. I recall having to do a similar clone to fix reading BytesWritable in this patch.
There's an additional wrinkle to allowing users to set batch size (which I previously raised in #755). Currently, we ignore batch size when checking if 2 serializers are equal and when they are considered equal, re-serialization is a no-op. If we expose batch size to users, we have to honor it. In that case, do we still want to keep current behavior of ```union```, where it doesn't proactively re-serialize if 2 rdd's serializers differ only in batch size? Adding to the confusion is current default batch size set in the SparkContext is 1000, which differs from the 10 that we pick for ```saveAsPickleFile``` and ```SchemaRDD.javaToPython```. I don't know the reason why 1000 was chosen for SparkContext. Was it for spark's own objects, not user objects? Would a single default batch size work for both cases? As long as we can get a consensus on these, I'm happy to sort things out either in this patch or a follow-up.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15152359
--- Diff: docs/programming-guide.md ---
@@ -403,31 +403,30 @@ PySpark SequenceFile support loads an RDD within Java, and pickles the resulting
<tr><td>BooleanWritable</td><td>bool</td></tr>
<tr><td>BytesWritable</td><td>bytearray</td></tr>
<tr><td>NullWritable</td><td>None</td></tr>
-<tr><td>ArrayWritable</td><td>list of primitives, or tuple of objects</td></tr>
--- End diff --
Ah, I see, it looks like in Scala we can write them but not read them. It's probably fine to remove them from the table then.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15601800
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -65,20 +66,52 @@ private[python] object SerDeUtil extends Logging {
* by PySpark. By default, if serialization fails, toString is called and the string
* representation is serialized
*/
- def rddToPython(rdd: RDD[(Any, Any)]): RDD[Array[Byte]] = {
+ def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
val (keyFailed, valueFailed) = checkPickle(rdd.first())
rdd.mapPartitions { iter =>
val pickle = new Pickler
- iter.map { case (k, v) =>
- if (keyFailed && valueFailed) {
- pickle.dumps(Array(k.toString, v.toString))
- } else if (keyFailed) {
- pickle.dumps(Array(k.toString, v))
- } else if (!keyFailed && valueFailed) {
- pickle.dumps(Array(k, v.toString))
+ val cleaned = iter.map { case (k, v) =>
+ val key = if (keyFailed) k.toString else k
+ val value = if (valueFailed) v.toString else v
+ Array[Any](key, value)
+ }
+ if (batchSize > 1) {
+ cleaned.grouped(batchSize).map(batched => pickle.dumps(seqAsJavaList(batched)))
+ } else {
+ cleaned.map(pickle.dumps(_))
+ }
+ }
+ }
+
+ /**
+ * Convert an RDD of serialized Python tuple (K, V) to RDD[(K, V)].
+ */
+ def pythonToPairRDD[K, V](pyRDD: RDD[Array[Byte]], batchSerialized: Boolean): RDD[(K, V)] = {
+ def isPair(obj: Any): Boolean = {
+ Option(obj.getClass.getComponentType).map(!_.isPrimitive).getOrElse(false) &&
+ obj.asInstanceOf[Array[_]].length == 2
+ }
+ pyRDD.mapPartitions { iter =>
+ val unpickle = new Unpickler
+ val unpickled =
--- End diff --
This `batchSerialized`-respecting unpickling logic should probably live in its own function so that it can also be used by `pythonToJavaMap`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50673853
I've merged this. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile and saveAs...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-48431572
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49810818
@pwendell I renamed file HBaseConverter.scala to HBaseConverters.scala. Now I failed Scala style checks. How can I fix it? Thx.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50206155
If you call `first()` on an RDD, we only perform as much computation as necessary to return the first result. If the RDD's lineage chain contains a shuffle stage, we have to perform that entire map-side of the shuffle stage before we can return the first result (e.g. if I call `someRDD.groupByKey().first()`). The map-side shuffle data should be automatically materialized to disk but I think we'd still pay the cost of re-computing the first partition of the post-shuffle stage when transforming the full RDD.
Here's a quick experiment in the shell (with `./bin/spark-shell --master "local-cluster[2,2,512]"`):
```scala
scala> val myRDD = sc.parallelize(1 to 10).flatMap(x => (1 to 10).map(y => (x, y)))
myRDD: org.apache.spark.rdd.RDD[(Int, Int)] = FlatMappedRDD[1] at flatMap at <console>:12
scala> val grouped = myRDD.groupByKey(4)
grouped: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = MappedValuesRDD[9] at groupByKey at <console>:14
scala> grouped.first()
14/07/25 14:14:09 INFO SparkContext: Starting job: first at <console>:17
14/07/25 14:14:09 INFO DAGScheduler: Registering RDD 1 (flatMap at <console>:12)
14/07/25 14:14:09 INFO DAGScheduler: Got job 3 (first at <console>:17) with 1 output partitions (allowLocal=true)
14/07/25 14:14:09 INFO DAGScheduler: Final stage: Stage 6(first at <console>:17)
14/07/25 14:14:09 INFO DAGScheduler: Parents of final stage: List(Stage 7)
14/07/25 14:14:09 INFO DAGScheduler: Missing parents: List(Stage 7)
14/07/25 14:14:09 INFO DAGScheduler: Submitting Stage 7 (FlatMappedRDD[1] at flatMap at <console>:12), which has no missing parents
14/07/25 14:14:09 INFO DAGScheduler: Submitting 4 missing tasks from Stage 7 (FlatMappedRDD[1] at flatMap at <console>:12)
// Run entire map side of the shuffle stage:
14/07/25 14:14:09 INFO TaskSchedulerImpl: Adding task set 7.0 with 4 tasks
14/07/25 14:14:09 INFO TaskSetManager: Starting task 0.0 in stage 7.0 (TID 15, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1717 bytes)
14/07/25 14:14:09 INFO TaskSetManager: Starting task 1.0 in stage 7.0 (TID 16, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1717 bytes)
14/07/25 14:14:09 INFO TaskSetManager: Starting task 2.0 in stage 7.0 (TID 17, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1717 bytes)
14/07/25 14:14:09 INFO TaskSetManager: Starting task 3.0 in stage 7.0 (TID 18, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1717 bytes)
14/07/25 14:14:09 INFO TaskSetManager: Finished task 2.0 in stage 7.0 (TID 17) in 14 ms on dhcp-47-206.eecs.berkeley.edu (1/4)
14/07/25 14:14:09 INFO TaskSetManager: Finished task 1.0 in stage 7.0 (TID 16) in 15 ms on dhcp-47-206.eecs.berkeley.edu (2/4)
14/07/25 14:14:09 INFO TaskSetManager: Finished task 0.0 in stage 7.0 (TID 15) in 17 ms on dhcp-47-206.eecs.berkeley.edu (3/4)
14/07/25 14:14:09 INFO TaskSetManager: Finished task 3.0 in stage 7.0 (TID 18) in 16 ms on dhcp-47-206.eecs.berkeley.edu (4/4)
14/07/25 14:14:09 INFO DAGScheduler: Stage 7 (flatMap at <console>:12) finished in 0.018 s
14/07/25 14:14:09 INFO DAGScheduler: looking for newly runnable stages
14/07/25 14:14:09 INFO TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks have all completed, from pool
14/07/25 14:14:09 INFO DAGScheduler: running: Set()
14/07/25 14:14:09 INFO DAGScheduler: waiting: Set(Stage 6)
14/07/25 14:14:09 INFO DAGScheduler: failed: Set()
14/07/25 14:14:09 INFO DAGScheduler: Missing parents for Stage 6: List()
14/07/25 14:14:09 INFO DAGScheduler: Submitting Stage 6 (MappedValuesRDD[9] at groupByKey at <console>:14), which is now runnable
// Only ran a single post-shuffle task:
14/07/25 14:14:09 INFO DAGScheduler: Submitting 1 missing tasks from Stage 6 (MappedValuesRDD[9] at groupByKey at <console>:14)
14/07/25 14:14:09 INFO TaskSchedulerImpl: Adding task set 6.0 with 1 tasks
14/07/25 14:14:09 INFO TaskSetManager: Starting task 0.0 in stage 6.0 (TID 19, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1721 bytes)
14/07/25 14:14:09 INFO MapOutputTrackerMasterActor: Asked to send map output locations for shuffle 3 to spark@dhcp-47-206.eecs.berkeley.edu:53826
14/07/25 14:14:09 INFO MapOutputTrackerMaster: Size of output statuses for shuffle 3 is 177 bytes
14/07/25 14:14:10 INFO TaskSetManager: Finished task 0.0 in stage 6.0 (TID 19) in 181 ms on dhcp-47-206.eecs.berkeley.edu (1/1)
14/07/25 14:14:10 INFO DAGScheduler: Stage 6 (first at <console>:17) finished in 0.182 s
14/07/25 14:14:10 INFO TaskSchedulerImpl: Removed TaskSet 6.0, whose tasks have all completed, from pool
14/07/25 14:14:10 INFO SparkContext: Job finished: first at <console>:17, took 0.208059 s
res6: (Int, Iterable[Int]) = (4,ArrayBuffer(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
// Demonstration that the map side of the shuffle was automatically cached / materialized:
scala> grouped.collect().size
14/07/25 14:16:03 INFO SparkContext: Starting job: collect at <console>:17
14/07/25 14:16:03 INFO DAGScheduler: Got job 5 (collect at <console>:17) with 4 output partitions (allowLocal=false)
14/07/25 14:16:03 INFO DAGScheduler: Final stage: Stage 10(collect at <console>:17)
14/07/25 14:16:03 INFO DAGScheduler: Parents of final stage: List(Stage 11)
14/07/25 14:16:03 INFO DAGScheduler: Missing parents: List()
14/07/25 14:16:03 INFO DAGScheduler: Submitting Stage 10 (MappedValuesRDD[9] at groupByKey at <console>:14), which has no missing parents
14/07/25 14:16:03 INFO DAGScheduler: Submitting 4 missing tasks from Stage 10 (MappedValuesRDD[9] at groupByKey at <console>:14)
14/07/25 14:16:03 INFO TaskSchedulerImpl: Adding task set 10.0 with 4 tasks
14/07/25 14:16:03 INFO TaskSetManager: Starting task 0.0 in stage 10.0 (TID 21, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1710 bytes)
14/07/25 14:16:03 INFO TaskSetManager: Starting task 1.0 in stage 10.0 (TID 22, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1710 bytes)
14/07/25 14:16:03 INFO TaskSetManager: Starting task 2.0 in stage 10.0 (TID 23, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1710 bytes)
14/07/25 14:16:03 INFO TaskSetManager: Starting task 3.0 in stage 10.0 (TID 24, dhcp-47-206.eecs.berkeley.edu, PROCESS_LOCAL, 1710 bytes)
14/07/25 14:16:03 INFO TaskSetManager: Finished task 2.0 in stage 10.0 (TID 23) in 26 ms on dhcp-47-206.eecs.berkeley.edu (1/4)
14/07/25 14:16:03 INFO TaskSetManager: Finished task 0.0 in stage 10.0 (TID 21) in 30 ms on dhcp-47-206.eecs.berkeley.edu (2/4)
14/07/25 14:16:03 INFO TaskSetManager: Finished task 1.0 in stage 10.0 (TID 22) in 32 ms on dhcp-47-206.eecs.berkeley.edu (3/4)
14/07/25 14:16:03 INFO TaskSetManager: Finished task 3.0 in stage 10.0 (TID 24) in 34 ms on dhcp-47-206.eecs.berkeley.edu (4/4)
14/07/25 14:16:03 INFO DAGScheduler: Stage 10 (collect at <console>:17) finished in 0.036 s
14/07/25 14:16:03 INFO TaskSchedulerImpl: Removed TaskSet 10.0, whose tasks have all completed, from pool
14/07/25 14:16:03 INFO SparkContext: Job finished: collect at <console>:17, took 0.040781 s
res8: Int = 10
```
In the worst case I think we only re-run one additional post-shuffle task. In fact, `first()` is implemented in terms of `take(1)`, which pushes the limit of `1` into the post-shuffle stage, so we probably won't run the full task anyways. So, I think it's fine to use `first()` here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15275404
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -31,13 +31,14 @@ import org.apache.spark.annotation.Experimental
* transformation code by overriding the convert method.
*/
@Experimental
-trait Converter[T, U] extends Serializable {
+trait Converter[T, +U] extends Serializable {
--- End diff --
(For better or worse, we don't really use it elsewhere in Spark)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50207653
@JoshRosen hi, thanks for your detailed explanation. I'm not sure I get your point or maybe I didn't make myself clear. I'm mainly concerned about the case of reading sequence files and was looking to see if I could unpersist the broadcasted Hadoop conf used by the read job. What I can gather from your note is that the answer is no, I can't unpersist the conf since the RDD may not have been fully read in after the RDD.first() call, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50108287
Uploaded a new patch adding batch serialization to Python when reading sequence files. For some reason, the test on custom class (Java Bean) no longer works and I disabled it. I suspect it's a bug in Pyrolite or Pickle.
Also addressed Josh's comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50223494
QA tests have started for PR 1338. This patch DID NOT merge cleanly! <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17216/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50109757
More on the JavaBean test failure, it seems for RDD[(key, Bean)], all the keys in a partition (or batch, I didn't test partition size > batch size) are paired with the last Bean in that partition (or batch) on the Python side. Does it ring any bells?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50387946
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17301/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-49198901
Great - I will review in more detail after that. Would be great to get this
merged before 1.1 freeze so PySpark I/O for inputformat and outputformat is
in for the next release!
On Tue, Jul 15, 2014 at 1:07 AM, kanzhang <no...@github.com> wrote:
> @MLnick <https://github.com/MLnick> I'll see if I can add couple output
> converter examples as well. Thx.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/1338#issuecomment-48971710>.
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50187471
One difference between this case and `saveAsPickleFile` is that here the actual on-disk output is independent of the batch size, since the data gets unbatched in Java, while `saveAsPickleFile` actually writes the batched data directly to a file.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50394525
QA results for PR 1338:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17301/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on a diff in the pull request:
https://github.com/apache/spark/pull/1338#discussion_r15431198
--- Diff: python/pyspark/rdd.py ---
@@ -964,6 +964,106 @@ def first(self):
"""
return self.take(1)[0]
+ def saveAsNewAPIHadoopDataset(self, conf, keyConverter=None, valueConverter=None):
+ """
+ Output a Python RDD of key-value pairs (of form C{RDD[(K, V)]}) to any Hadoop file
+ system, using the new Hadoop OutputFormat API (mapreduce package). Keys/values are
+ converted for output using either user specified converters or, by default,
+ L{org.apache.spark.api.python.JavaToWritableConverter}.
+
+ @param conf: Hadoop job configuration, passed in as a dict
+ @param keyConverter: (None by default)
+ @param valueConverter: (None by default)
+ """
+ jconf = self.ctx._dictToJavaMap(conf)
+ reserialized = self._reserialize(BatchedSerializer(PickleSerializer(), 10))
--- End diff --
@JoshRosen @MLnick the batch size here only affects transient data when writing and the re-serialization shouldn't be done if the data is already in pickle format (batch serialized or not). I'm uploading a patch to that effect. Since the batch size has no effect on the data persisted in files, I'm not exposing it to users. See you have any further comments and if you have a better suggestion for the default size 10, let me know.
@mateiz what you suggested is exposing the batch size for reading. I'll work on that next.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50547331
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17386/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50223574
Updated the patch to allow users to specify batch size when reading in a sequence file. When batch size is 1, the pickled data is unbatched (and custom Writables are not cloned). Default batch size is set to 10. Added a few tests for unbatched cases. Let me know if you have any further comments. Thanks everyone for review and suggestions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request: [SPARK-2024] Add saveAsSequenceFile to PySpark
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1338#issuecomment-50576604
QA tests have started for PR 1338. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17424/consoleFull
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---