You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by MLnick <gi...@git.apache.org> on 2014/04/19 21:37:54 UTC
[GitHub] spark pull request: [WIP] SPARK-1416: PySpark support for Sequence...
GitHub user MLnick opened a pull request:
https://github.com/apache/spark/pull/455
[WIP] SPARK-1416: PySpark support for SequenceFile and Hadoop InputFormats
So I finally resurrected this PR. It seems the old one against the incubator mirror is no longer available, so I cannot reference it.
This adds initial support for reading Hadoop ```SequenceFile```s, as well as arbitrary Hadoop ```InputFormat```s, in PySpark.
# Overview
The basics are as follows:
1. ```PythonRDD``` object contains the relevant methods, that are in turn invoked by ```SparkContext``` in PySpark
2. The SequenceFile or InputFormat is read on the Scala side and converted from ```Writable``` instances to the relevant Scala classes (in the case of primitives)
3. ```msgpack``` is used to serialize Scala objects. An attempt is made to register custom classes with ```msgpack```. If this fails, the fallback is ```toString```
4. ```MsgpackSerializer``` on the Python side deserializes using ```msgpack-python```.
This works "out the box" for simple ```Writable```s:
* ```Text```
* ```IntWritable```, ```DoubleWritable```
* ```NullWritable```
* ```BooleanWritable```
* ```BytesWritable```
* ```MapWritable```
It also works for simple, "struct-like" classes, with fields and a no-arg constructor.
I've tested it out with ```ESInputFormat``` and it works very nicely:
```python
conf = {"es.resource" : "index/type" }
rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat", "org.apache.hadoop.io.NullWritable", "
org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
rdd.first()
```
I suspect for things like HBase/Cassandra it will be a bit trickier to get it to work out the box.
# Some things still outstanding:
1. Requires ```msgpack-python``` and will fail without it. As originally discussed with Josh, add a ```as_strings``` argument that defaults to ```False```, that can be used if ```msgpack-python``` is not available
2. I see from https://github.com/apache/spark/pull/363 that Pyrolite is being used there for SerDe between Scala and Python. @ahirreddy @mateiz what is the plan behind this - is Pyrolite preferred? It seems from a cursory glance that adapting the ```msgpack```-based SerDe here to use Pyrolite wouldn't be too hard
3. Support the key and value "wrapper" that would allow a Scala/Java function to be plugged in that would transform whatever the key/value Writable class is into something that can be serialized (e.g. convert some custom Writable to a case class or map that can be easily serialized)
4. Support ```saveAsSequenceFile``` and ```saveAsHadoopFile``` etc. This would require SerDe in the reverse direction if it should be compatible with Java too. Perhaps Pyrolite would make this quite simple?
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/MLnick/spark-1 pyspark-inputformats
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/455.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 #455
----
commit d86325f394e536c6b2d5cb86d08e35d508fb23d7
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-09T06:53:11Z
Initial WIP of PySpark support for SequenceFile and arbitrary Hadoop InputFormat
commit 4b0a43fa2236ed997a5f9bd273151f21c1cc23f8
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-12T10:31:18Z
Refactoring utils into own objects. Cleaning up old commented-out code
commit c304cc8ae1a39b3c58f21153e7311cfab7dfac9e
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-15T06:32:37Z
Adding supporting sequncefiles for tests. Cleaning up
commit 4e7c9e359bbbb66fd658270af266b406ffedfa05
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-15T09:29:16Z
Merge remote-tracking branch 'upstream/master' into pyspark-inputformats
Conflicts:
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
project/SparkBuild.scala
commit 818a1e6fd6c25a689629358f540de54d32f9acbf
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-15T14:32:45Z
Add seqencefile and Hadoop InputFormat support to PythonRDD
commit 4294cbb9ea89fdcb61993bbf903e6ec157b08164
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-19T09:06:04Z
Add old Hadoop api methods. Clean up and expand comments. Clean up argument names
commit 0f5cd844fe63b7dd9a3e0a920f9a26a25b9f5d1a
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-19T09:06:37Z
Remove unused pair UTF8 class. Add comments to msgpack deserializer
commit f1d73e3cad7ba7db176ed7cb48c901318e732266
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-19T09:07:22Z
mergeConfs returns a copy rather than mutating one of the input arguments
commit 4d7ef2ecc4de91daae4f0de9724930b7f32dc96a
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-19T09:08:38Z
Fix indentation
commit eb400366ee7617c7c92682d49f64079092d94031
Author: Nick Pentreath <ni...@gmail.com>
Date: 2013-12-19T11:48:40Z
Remove unused comment lines
commit 1c8efbc4a87cd7d719d8fef4e41781c67b414a6f
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-01-13T11:45:31Z
Merge remote-tracking branch 'upstream/master' into pyspark-inputformats
Conflicts:
project/SparkBuild.scala
python/pyspark/context.py
commit 619c0fa2452829882628ed2c5cee06f1dae68f6f
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-01-20T08:49:02Z
Merge remote-tracking branch 'upstream/master' into pyspark-inputformats
Conflicts:
project/SparkBuild.scala
commit 703ee65c5feb402f798c7f820466fef8fd78334a
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-01-20T09:37:57Z
Add back msgpack
commit 174f52011ecfa8b6d7cb37592d5b0ea2bca9e9b5
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-01-20T09:54:26Z
Add back graphx settings
commit 795a763a39db9830d4f1917147b076ec46b6d238
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-01-20T12:14:01Z
Change name to WriteInputFormatTestDataGenerator. Cleanup some var names. Use SPARK_HOME in path for writing test sequencefile data.
commit 2beeedb6afccdf4a6d27cc7550a82ec581a0f54a
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-02-08T15:29:34Z
Merge remote-tracking branch 'upstream/master' into pyspark-inputformats
Conflicts:
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
python/pyspark/context.py
commit 97ef708a4730bc79cccd56d1106eeddb474d58fc
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-02-14T16:27:02Z
Remove old writeToStream
commit 41856a51666a0ed94dc7063cf33a459b0352e302
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-03-19T08:00:17Z
Merge branch 'master' into pyspark-inputformats
Conflicts:
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
commit f2d76a0c18fda6204e511d8b673b5adc749e67f1
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-03-19T08:10:26Z
Merge branch 'master' into pyspark-inputformats
Conflicts:
project/SparkBuild.scala
commit e67212a6e3fdf13930fbb6e2b8c328eb371c237b
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-03-19T08:11:01Z
Add back msgpack dependency
commit dd579220dc2d41b546f3a82f8bd5d1b82a2a19cf
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-10T06:47:20Z
Merge remote-tracking branch 'upstream/master' into pyspark-inputformats
Conflicts:
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
project/SparkBuild.scala
python/pyspark/context.py
python/pyspark/serializers.py
commit d72bf18c92589f22335f1119e702d88f263a3145
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-10T07:45:37Z
msgpack
commit 0c612e5c56a1a9ff33535be69242b6cd42774c70
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-12T13:32:52Z
Merge branch 'master' into pyspark-inputformats
commit 65360d59247808db0c7db260c34b477e3cfb6b56
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T13:41:36Z
Adding test SequenceFiles
commit 25da1ca8dd2cf8117342c322ee86f9afd2e52f29
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T13:42:39Z
Add generator for nulls, bools, bytes and maps
commit 7237263b3f23591c1969df475641aac65bb9ecfe
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T13:44:08Z
Add back msgpack serializer and hadoop file code lost during merging
commit a67dfadc02bd0fde78a3de76d03a5e26a870dd44
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T13:44:46Z
Clean up Msgpack serialization and registering
commit 1bbbfb07c3a3da2c06a0f491eddb864eb9dc79f0
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T13:45:22Z
Clean up SparkBuild from merge
commit 9d2256e06e4e63eed47791f049f4702b8c55773d
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T14:18:38Z
Merge branch 'master' into pyspark-inputformats
Conflicts:
core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
project/SparkBuild.scala
commit f6aac55ef7a78569e79e9ee5b5582ee1f2ce56d5
Author: Nick Pentreath <ni...@gmail.com>
Date: 2014-04-18T14:19:17Z
Bring back msgpack
----
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148169
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import scala.util.Try
+import org.apache.spark.rdd.RDD
+import org.apache.spark.Logging
+import scala.util.Success
+import scala.util.Failure
+import net.razorvine.pickle.Pickler
+
+/**
+ * Utilities for serialization / deserialization between Python and Java, using MsgPack.
+ * Also contains utilities for converting [[org.apache.hadoop.io.Writable]] ->
+ * Scala objects and primitives
+ */
+private[python] object SerDeUtil extends Logging {
+
+ /**
+ * Convert an RDD of key-value pairs to an RDD of serialized Python objects, that is usable
+ * by PySpark. By default, if serialization fails, toString is called and the string
+ * representation is serialized
+ */
+ def rddToPython[K, V](rdd: RDD[(K, V)]): RDD[Array[Byte]] = {
+ rdd.mapPartitions{ iter =>
--- End diff --
Add spaces before { here 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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40955747
Right that makes sense.
Once I push the Pyrolite changes I will look into that
Sent from my iPhone
> On 21 Apr 2014, at 19:28, Matei Zaharia <no...@github.com> wrote:
>
> Yeah, it would be good to generate them during the test and tear them down after (you can just call the Scala class that generates them from Python). It's kind of confusing to include them in git.
>
> —
> Reply to this email directly or view it on GitHub.
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13521516
--- Diff: docs/programming-guide.md ---
@@ -378,11 +378,88 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
-</div>
+### SequenceFile and Hadoop InputFormats
+
+In addition to reading text files, PySpark supports reading [SequenceFile](http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html)
+and any arbitrary [InputFormat](http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/InputFormat.html).
+
+#### Writable Support
+
+PySpark SequenceFile support loads an RDD within Java, and pickles the resulting Java objects using
+[Pyrolite](https://github.com/irmen/Pyrolite/). The following Writables are automatically converted:
+
+<table class="table">
+<tr><th>Writable Type</th><th>Python Type</th></tr>
+<tr><td>Text</td><td>unicode str</td></tr>
+<tr><td>IntWritable</td><td>int</td></tr>
+<tr><td>FloatWritable</td><td>float</td></tr>
+<tr><td>DoubleWritable</td><td>float</td></tr>
+<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>
+<tr><td>MapWritable</td><td>dict</td></tr>
+<tr><td>Custom Class conforming to Java Bean conventions</td>
+ <td>dict of public properties (via JavaBean getters and setters) + __class__ for the class type</td></tr>
+</table>
+
+#### Loading SequenceFiles
+
+Similarly to text files, SequenceFiles can be loaded by specifying the path. The key and value
+classes can be specified, but for standard Writables it should work without requiring this.
+
+{% highlight python %}
+>>> rdd = sc.sequenceFile("path/to/sequencefile/of/doubles")
+>>> rdd.collect() # this example has DoubleWritable keys and Text values
+[(1.0, u'aa'),
+ (2.0, u'bb'),
+ (2.0, u'aa'),
+ (3.0, u'cc'),
+ (2.0, u'bb'),
+ (1.0, u'aa')]
+{% endhighlight %}
+
+#### Loading Arbitrary Hadoop InputFormats
+
+PySpark can also read any Hadoop InputFormat, for both 'new' and 'old' Hadoop APIs. If required,
+a Hadoop configuration can be passed in as a Python dict. Here is an example using the
+Elasticsearch ESInputFormat:
+
+{% highlight python %}
+$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
+>>> conf = {"es.resource" : "index/type"} # assume Elasticsearch is running on localhost defaults
+>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",\
+ "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
+>>> rdd.first() # the result is a MapWritable that is converted to a Python dict
+(u'Elasticsearch ID',
+ {u'field1': True,
+ u'field2': u'Some Text',
+ u'field3': 12345})
+{% endhighlight %}
+Note that, if the InputFormat simply depends on a Hadoop configuration and/or input path, and
+the key and value classes can easily be converted according to the above table,
+then this approach should work well for such cases.
+
+If you have custom serialized binary data (like pulling data from Cassandra / HBase) or custom
+classes that don't conform to the JavaBean requirements, then you will first need to
+transform that data on the Scala/Java side to something which can be handled by Pyrolite's pickler.
+A [Converter](api/scala/index.html#org.apache.spark.api.python.Converter) trait is provided
+for this. Simply extend this trait and implement your transformation code in the ```convert```
+method. The ensure this class is packaged into your Spark job jar and included on the PySpark
--- End diff --
Typo: "The ensure".
Also, I'd mention that they need to include both the `Convereter` implementation and also any dependencies required to read from the storage system (e.g. `hbase-client`) those actually aren't packaged by default with Spark except in the examples.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46767642
Thanks a ton! One thing - how can I pull spark core 1.1 from maven?
[ERROR] Failed to execute goal on project avro: Could not resolve
dependencies for project example:avro:jar:0.1: Could not find artifact
org.apache.spark:spark-core_2.10:jar:1.1.0-SNAPSHOT in scala-tools.org (
http://scala-tools.org/repo-releases) -> [Help 1]
On Fri, Jun 20, 2014 at 10:45 PM, MLnick <no...@github.com> wrote:
> @rjurney <https://github.com/rjurney> this works for me (building Spark
> from current master): https://gist.github.com/MLnick/5864741781b9340cb211
>
> if you run mvn package and then add that to SPARK_CLASSPATH and use it in
> IPython console.
>
> However it seems to come through as only strings (not a dict). I verified
> that if I take only the string field and explicitly convert to string (ie Map[String,
> String]) then it works. I suspect then that Avro doesn't have the type
> information at all, so Pyrolite cannot pickle it. I guess you might have to
> do something more in depth in the AvroConverter to read the type info
> from the Avro schema and do a cast...
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/455#issuecomment-46745394>.
>
--
Russell Jurney twitter.com/rjurney russell.jurney@gmail.com datasyndrome.com
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45437963
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41132798
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13428056
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
--- End diff --
Making this change I end up with:
```scala
private[python] class ConverterRegistry extends Logging {
var keyConverter: Converter[Any, Any] = DefaultConverter
var valueConverter: Converter[Any, Any] = DefaultConverter
...
private def register(converterClass: String): Converter[Any, Any] = {
Try {
val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
converter
} match {
case Success(s) => s
case Failure(err) =>
logError(s"Failed to register converter: $converterClass")
throw err
}
}
```
Is this the best approach or is there another better way?
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r11919174
--- Diff: python/pyspark/serializers.py ---
@@ -114,11 +113,6 @@ class FramedSerializer(Serializer):
where C{length} is a 32-bit integer and data is C{length} bytes.
"""
- def __init__(self):
- # On Python 2.6, we can't write bytearrays to streams, so we need to convert them
- # to strings first. Check if the version number is that old.
- self._only_write_strings = sys.version_info[0:2] <= (2, 6)
--- End diff --
Actually all the changes in that file have this problem, they should all be 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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41189494
Yeah, let me look through this -- it may be too big a change to add to 1.0 now, though it may be worth adding it as an experimental API. In the worst case though, 1.1 will probably be 3 months after, so it's not too bad.
The test that failed is currently flaky, it might not be due to your changes.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45411714
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45099077
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45361465
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878965
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46773577
1.1 is not released yet. This PR is in master but not in 1.0 (it may be released in 1.0.1 or if not then 1.1).
So you'll have to clone master and run sbt/sbt publish-local which will publish the maven and sbt artifacts to your local repos.
—
Sent from Mailbox
On Sun, Jun 22, 2014 at 1:22 AM, Russell Jurney <no...@github.com>
wrote:
> Thanks a ton! One thing - how can I pull spark core 1.1 from maven?
> [ERROR] Failed to execute goal on project avro: Could not resolve
> dependencies for project example:avro:jar:0.1: Could not find artifact
> org.apache.spark:spark-core_2.10:jar:1.1.0-SNAPSHOT in scala-tools.org (
> http://scala-tools.org/repo-releases) -> [Help 1]
> On Fri, Jun 20, 2014 at 10:45 PM, MLnick <no...@github.com> wrote:
>> @rjurney <https://github.com/rjurney> this works for me (building Spark
>> from current master): https://gist.github.com/MLnick/5864741781b9340cb211
>>
>> if you run mvn package and then add that to SPARK_CLASSPATH and use it in
>> IPython console.
>>
>> However it seems to come through as only strings (not a dict). I verified
>> that if I take only the string field and explicitly convert to string (ie Map[String,
>> String]) then it works. I suspect then that Avro doesn't have the type
>> information at all, so Pyrolite cannot pickle it. I guess you might have to
>> do something more in depth in the AvroConverter to read the type info
>> from the Avro schema and do a cast...
>>
>> —
>> Reply to this email directly or view it on GitHub
>> <https://github.com/apache/spark/pull/455#issuecomment-46745394>.
>>
> --
> Russell Jurney twitter.com/rjurney russell.jurney@gmail.com datasyndrome.com
> ---
> Reply to this email directly or view it on GitHub:
> https://github.com/apache/spark/pull/455#issuecomment-46767642
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148051
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkContext
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+
+
+/**
+ * Utilities for working with Python objects -> Hadoop-related objects
+ */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a Map of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]) = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration) = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)]) = {
+ rdd.map{
--- End diff --
Space before {
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49684527
@ericgarcia This is awesome. How can I test this code out? I can handle patching trunk, but how do I call the converter?
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r11919199
--- Diff: python/pyspark/context.py ---
@@ -423,11 +514,8 @@ def _getJavaStorageLevel(self, storageLevel):
raise Exception("storageLevel must be of type pyspark.StorageLevel")
newStorageLevel = self._jvm.org.apache.spark.storage.StorageLevel
- return newStorageLevel(storageLevel.useDisk,
- storageLevel.useMemory,
- storageLevel.useOffHeap,
- storageLevel.deserialized,
- storageLevel.replication)
+ return newStorageLevel(storageLevel.useDisk, storageLevel.useMemory,
+ storageLevel.deserialized, storageLevel.replication)
--- End diff --
This is also reverting an earlier change and will break; you need to keep it the way it was.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45096021
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41208393
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14386/
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50424644
Ok, so now I rebuilt with my specific CDH version, and I get this when I run ./sbin/start-master.sh:
Spark Command: /usr/java/jdk1.8.0//bin/java -cp ::/home/hivedata/spark/conf:/home/hivedata/spark/assembly/target/scala-2.10/spark-assembly-1.1.0-SNAPSHOT-hadoop2.0.0-mr1-cdh4.4.0.jar -XX:MaxPermSize=128m -Dspark.akka.logLifecycleEvents=true -Xms512m -Xmx512m org.apache.spark.deploy.master.Master --ip hivecluster2 --port 7077 --webui-port 8080
========================================
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0
14/07/28 18:33:37 INFO Master: Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
14/07/28 18:33:37 INFO Master: Registered signal handlers for [TERM, HUP, INT]
14/07/28 18:33:37 INFO SecurityManager: Changing view acls to: hivedata
14/07/28 18:33:37 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hivedata)
14/07/28 18:33:38 INFO Slf4jLogger: Slf4jLogger started
14/07/28 18:33:38 INFO Remoting: Starting remoting
14/07/28 18:33:38 ERROR ActorSystemImpl: Uncaught fatal error from thread [sparkMaster-akka.actor.default-dispatcher-2] shutting down ActorSystem [sparkMaster]
java.lang.VerifyError: (class: org/jboss/netty/channel/socket/nio/NioWorkerPool, method: createWorker signature: (Ljava/util/concurrent/Executor;)Lorg/jboss/netty/channel/socket/nio/AbstractNioWorker;) Wrong return type in function
at akka.remote.transport.netty.NettyTransport.<init>(NettyTransport.scala:282)
at akka.remote.transport.netty.NettyTransport.<init>(NettyTransport.scala:239)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:408)
at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$2.apply(DynamicAccess.scala:78)
at scala.util.Try$.apply(Try.scala:161)
at akka.actor.ReflectiveDynamicAccess.createInstanceFor(DynamicAccess.scala:73)
at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$3.apply(DynamicAccess.scala:84)
at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$3.apply(DynamicAccess.scala:84)
at scala.util.Success.flatMap(Try.scala:200)
at akka.actor.ReflectiveDynamicAccess.createInstanceFor(DynamicAccess.scala:84)
at akka.remote.EndpointManager$$anonfun$8.apply(Remoting.scala:618)
at akka.remote.EndpointManager$$anonfun$8.apply(Remoting.scala:610)
at scala.collection.TraversableLike$WithFilter$$anonfun$map$2.apply(TraversableLike.scala:722)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$WithFilter.map(TraversableLike.scala:721)
at akka.remote.EndpointManager.akka$remote$EndpointManager$$listens(Remoting.scala:610)
at akka.remote.EndpointManager$$anonfun$receive$2.applyOrElse(Remoting.scala:450)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
14/07/28 18:33:38 INFO RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon.
14/07/28 18:33:38 INFO RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports.
14/07/28 18:33:38 INFO RemoteActorRefProvider$RemotingTerminator: Remoting shut down.
Exception in thread "main" java.util.concurrent.TimeoutException: Futures timed out after [10000 milliseconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at akka.remote.Remoting.start(Remoting.scala:173)
at akka.remote.RemoteActorRefProvider.init(RemoteActorRefProvider.scala:184)
at akka.actor.ActorSystemImpl._start$lzycompute(ActorSystem.scala:579)
at akka.actor.ActorSystemImpl._start(ActorSystem.scala:577)
at akka.actor.ActorSystemImpl.start(ActorSystem.scala:588)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:111)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:104)
at org.apache.spark.util.AkkaUtils$.createActorSystem(AkkaUtils.scala:104)
at org.apache.spark.deploy.master.Master$.startSystemAndActor(Master.scala:808)
at org.apache.spark.deploy.master.Master$.main(Master.scala:788)
at org.apache.spark.deploy.master.Master.main(Master.scala)
---
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-1416: PySpark support for SequenceFile a...
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45687030
@MLnick hey Nick, are you planning on implementing the reverse direction - ```saveAsSequenceFile``` and ```saveAsHadoopFile```? If not, I'd like to give it a shot.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40973858
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-1416: PySpark support for SequenceFile a...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/455
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41209895
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45437958
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13414912
--- Diff: docs/programming-guide.md ---
@@ -378,9 +378,82 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+## SequenceFile and Hadoop InputFormats
+
+In addition to reading text files, PySpark supports reading [SequenceFile](http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html)
+and any arbitrary [InputFormat](http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/InputFormat.html).
+
+### Writable Support
+
+PySpark SequenceFile support loads an RDD within Java, and pickles the resulting Java objects using
+[Pyrolite](https://github.com/irmen/Pyrolite/). The following Writables are automatically converted:
+
+<table class="table">
+<tr><th>Writable Type</th><th>Scala Type</th><th>Python Type</th></tr>
+<tr><td>Text</td><td>String</td><td>unicode str</td></tr>
+<tr><td>IntWritable</td><td>Int</td><td>int</td></tr>
+<tr><td>FloatWritable</td><td>Float</td><td>float</td></tr>
+<tr><td>DoubleWritable</td><td>Double</td><td>float</td></tr>
+<tr><td>BooleanWritable</td><td>Boolean</td><td>bool</td></tr>
+<tr><td>BytesWritable</td><td>Array[Byte]</td><td>bytearray</td></tr>
+<tr><td>NullWritable</td><td>null</td><td>None</td></tr>
+<tr><td>ArrayWritable</td><td>Array[T]</td><td>list of primitives, or tuple of objects</td></tr>
+<tr><td>MapWritable</td><td>java.util.Map[K, V]</td><td>dict</td></tr>
+<tr><td>Custom Class</td><td>Custom Class conforming to Java Bean conventions</td>
+ <td>dict of public properties (via JavaBean getters and setters) + __class__ for the class type</td></tr>
+</table>
+
+### Loading SequenceFiles
+
+Similarly to text files, SequenceFiles can be loaded by specifying the path. The key and value
+classes can be specified, but for standard Writables it should work without requiring this.
+
+{% highlight python %}
+>>> rdd = sc.sequenceFile("path/to/sequencefile/of/doubles")
+>>> rdd.collect() # this example has DoubleWritable keys and Text values
+[(1.0, u'aa'),
+ (2.0, u'bb'),
+ (2.0, u'aa'),
+ (3.0, u'cc'),
+ (2.0, u'bb'),
+ (1.0, u'aa')]
+>>> help(sc.sequenceFile) # Show sequencefile documentation
+{% endhighlight %}
+
+### Loading Arbitrary Hadoop InputFormats
+
+PySpark can also read any Hadoop InputFormat, for both 'new' and 'old' Hadoop APIs. If required,
+a Hadoop configuration can be passed in as a Python dict. Here is an example using the
+Elasticsearch ESInputFormat:
+
+{% highlight python %}
+$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
+>>> conf = {"es.resource" : "index/type"} # assume Elasticsearch is running on localhost defaults
+>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",\
+ "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
+>>> rdd.first() # the result is a MapWritable that is converted to a Python dict
+(u'Elasticsearch ID',
+ {u'field1': True,
+ u'field2': u'Some Text',
+ u'field3': 12345})
+>>> help(sc.newAPIHadoopRDD) # Show help for new API Hadoop RDD
+{% endhighlight %}
+
+Note that, if the InputFormat simply depends on a Hadoop configuration and/or input path, and
+the key and value classes can easily be converted according to the above table,
+then this approach should work well for such cases.
+
+If you have custom serialized binary data (like pulling data from Cassandra / HBase) or custom
+classes that don't conform to the JavaBean requirements, then you will probably have to first
+transform that data on the Scala/Java side to something which can be handled by Pyrolite's pickler.
+
+Future support for custom 'converter' functions for keys/values that allows this to be written in Java/Scala,
+and called from Python, as well as support for writing data out as SequenceFileOutputFormat
+and other OutputFormats, is forthcoming.
--- End diff --
Is this still true or can we link to the Converter interface 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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45412792
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15526/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40939971
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14283/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40882224
Cool, thanks for porting this over! A few notes:
- I looked at msgpack in the past and one problem with it was that users need to install it separately through "pip" to use PySpark. Before this, we had no external package dependencies except NumPy for ML. For this reason it would be good to investigate Pyrolite instead (which just uses pickling on the Python side). If that doesn't work, we should write the code in a way that imports msgpack only if you're using one of these methods.
- There are a bunch of binary test files included, would it be possible to generate those programmatically instead (e.g. through saveAsSequenceFile, or through a Java-side static method)?
- The build is failing due to the scalastyle checker; you can run sbt scalastyle locally to do the same tests there. (See https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14259/console for the current errors).
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50515663
@ericgarcia Could you please create a public branch with this code in a working state and push it to your clone of spark so I can use that? I'm bad at merging conflicts.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45101260
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40937302
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40879000
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41189533
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147535
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
--- End diff --
The first parameter should be called `path` instead of `name` (unfortunately parameter names matter in Python due to the ability to pass named arguments).
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45439535
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15544/
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50422135
@JoshRosen
Huh, I'm going by http://spark.apache.org/docs/latest/hadoop-third-party-distributions.html
and I get:
sbt.ResolveException: unresolved dependency: org.apache.hadoop#hadoop-client;2.0.0-mr1-cdh4.X.X: not found
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50425084
Hmmm, I didn't clean before rebuilding with CDH 4.4. Trying that 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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45421855
@mateiz I see - if this is required for supporting arbitrary input types, I guess it's the only way. I'm hoping in the future though most people don't have to write their own converters because we support the most common storage layers. So I'll withdraw my concern.
@MLnick thanks for responding to the earlier feedback I think the patch is looking good. I made two minor comments, if you can address those than it's good with me.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-43988938
Merged build finished. All automated tests passed.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-51220911
It will be in release 1.1.
You should be able to check out branch-1.1 and build from source and it should work ok.
Otherwise 1.1 should be released within a couple weeks
—
Sent from Mailbox
On Tue, Aug 5, 2014 at 4:15 PM, Russell Jurney <no...@github.com>
wrote:
> @ericgarcia @srowen @MLnick
> Unfortunately when I follow those directions, I still get errors. It looks like I'll have to wait to get this functionality until its merged. Anyone have any idea when that will be? Is there a release in mind that this would be a part of?
> ---
> Reply to this email directly or view it on GitHub:
> https://github.com/apache/spark/pull/455#issuecomment-51202880
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: SPARK-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13513826
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
+
+ var keyConverter: Converter[Any, Any] = DefaultConverter
+ var valueConverter: Converter[Any, Any] = DefaultConverter
+
+ def convertKey(obj: Any): Any = keyConverter.convert(obj)
+
+ def convertValue(obj: Any): Any = valueConverter.convert(obj)
+
+ def registerKeyConverter(converterClass: String) = {
+ keyConverter = register(converterClass)
+ logInfo(s"Loaded and registered key converter ($converterClass)")
+ }
+
+ def registerValueConverter(converterClass: String) = {
+ valueConverter = register(converterClass)
+ logInfo(s"Loaded and registered value converter ($converterClass)")
+ }
+
+ private def register(converterClass: String): Converter[Any, Any] = {
+ Try {
+ val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
+ converter
+ } match {
+ case Success(s) => s
+ case Failure(err) =>
+ logError(s"Failed to register converter: $converterClass")
+ throw err
+ }
+ }
+}
+
+/** Utilities for working with Python objects -> Hadoop-related objects */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a [[java.util.Map]] of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]): Configuration = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration): Configuration = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)],
+ keyClass: String,
+ keyConverter: Option[String],
+ valueClass: String,
+ valueConverter: Option[String]) = {
+ rdd.mapPartitions { case iter =>
+ val registry = new ConverterRegistry
+ keyConverter.foreach(registry.registerKeyConverter(_))
+ valueConverter.foreach(registry.registerValueConverter(_))
--- End diff --
This will already create a single instance of the key and value converter for the partition.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41209887
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13472864
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
--- End diff --
That seems okay; are you worried that you have to cast it? You can cast any Converter to `Converter[Any, Any]` safely due to type erasure.
---
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-1416: PySpark support for SequenceFile a...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50451633
@rjurney This means you have two conflicting versions of Netty in the build. It may or may not be to do with the JIRA you cite, just because there are lots of Netties floating around and lots of ways they can collide. Maybe you can restate what version you are buildling, how you are building at this point and any modifications to the build? `mvn dependency:tree` will reveal where Netties are coming from.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50101315
I got this to run and I'm able to get work done!
Does this code have to be run on the latest Spark code? Would it run on 1.0?
On Tuesday, July 22, 2014, Eric Garcia <no...@github.com> wrote:
> @MLnick <https://github.com/MLnick>, I made a PR here: #1536
> <https://github.com/apache/spark/pull/1536>
> @rjurney <https://github.com/rjurney>, the updated code works for the
> .avro file you posted though it is still not fully implemented for *all*
> data types. Note that any null values in your data will show up as an empty
> string "". For some reason I could not get Java null to convert to Python
> None.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/455#issuecomment-49792170>.
>
--
Russell Jurney twitter.com/rjurney russell.jurney@gmail.com datasyndrome.com
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45101254
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415225
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
+ def convert(obj: Any): Any
+}
+
+object DefaultConverter extends Converter {
--- End diff --
Should be `private[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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45365821
Merged build finished. All automated tests passed.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13414845
--- Diff: docs/programming-guide.md ---
@@ -378,9 +378,82 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+## SequenceFile and Hadoop InputFormats
+
+In addition to reading text files, PySpark supports reading [SequenceFile](http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html)
+and any arbitrary [InputFormat](http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/InputFormat.html).
+
+### Writable Support
+
+PySpark SequenceFile support loads an RDD within Java, and pickles the resulting Java objects using
+[Pyrolite](https://github.com/irmen/Pyrolite/). The following Writables are automatically converted:
+
+<table class="table">
+<tr><th>Writable Type</th><th>Scala Type</th><th>Python Type</th></tr>
+<tr><td>Text</td><td>String</td><td>unicode str</td></tr>
+<tr><td>IntWritable</td><td>Int</td><td>int</td></tr>
+<tr><td>FloatWritable</td><td>Float</td><td>float</td></tr>
+<tr><td>DoubleWritable</td><td>Double</td><td>float</td></tr>
+<tr><td>BooleanWritable</td><td>Boolean</td><td>bool</td></tr>
+<tr><td>BytesWritable</td><td>Array[Byte]</td><td>bytearray</td></tr>
+<tr><td>NullWritable</td><td>null</td><td>None</td></tr>
+<tr><td>ArrayWritable</td><td>Array[T]</td><td>list of primitives, or tuple of objects</td></tr>
+<tr><td>MapWritable</td><td>java.util.Map[K, V]</td><td>dict</td></tr>
+<tr><td>Custom Class</td><td>Custom Class conforming to Java Bean conventions</td>
+ <td>dict of public properties (via JavaBean getters and setters) + __class__ for the class type</td></tr>
+</table>
--- End diff --
Does this table need a "Scala type" column? It seems like for anything except the last one, you can add just the Python type, and for the "custom class" you can make the left column be "custom class conforming to Java Bean conventions"
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45411654
@pwendell @mateiz ok I addressed those comments, and simplified things quite a bit. I removed the registry and just used ```.get()``` (etc) directly for the standard ```Writable``` types.
Also created factory method for instantiating a ```Converter``` and did that early on in ```PythonRDD``` methods.
Finally, used ```RDD.first()``` to check for Pickle-ability of the keys and values.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50514645
@ericgarcia @srowen Sorry, but again I can't make things go. I try to pull that request to branch-1.0.0 via: 'git fetch origin pull/455/head:master' but I get this:
From github.com:apache/spark
! [rejected] refs/pull/455/head -> master (non-fast-forward)
* [new tag] v0.9.2 -> v0.9.2
* [new tag] v1.0.2-rc1 -> v1.0.2-rc1
Russells-MacBook-Pro:spark-trunk rjurney$ git status
Which looks like it isn't applying?
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40937314
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45160638
CC @ahirreddy, @joshrosen, @holdenk for 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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45094108
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-1416: PySpark support for SequenceFile a...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50421644
@rjurney
Did you build Spark against your cluster's Hadoop version? Do you mind posting the build command that you used?
> Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.hadoopFile.
: java.lang.ClassCastException: org.apache.avro.mapreduce.AvroKeyInputFormat cannot be cast to org.apache.hadoop.mapred.InputFormat
You need to use `newAPIHadoopFile`, not `hadoopFile`, since `mapreduce.InputFormat`s can only be used with that API.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45439532
Merged build finished. All automated tests passed.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45432064
@pwendell thanks for the comments. Will make those amendments.
I get that SparkSQL will be the preferred way of doing things for structured data in general, so happy to mark this feature as advanced and experimental, and have it superseded by the relevant SparkSQL functionality later.
Having said this, I see this as simply adding missing API functionality to PySpark. All it does is bring PySpark on a par with Scala/Java in terms of being able to read from any ```InputFormat```. The vast majority of Spark users end up using ```textFile```, with others probably using ```SequenceFile``` and other common formats (such as HBase, Cassandra, Parquet etc). I expect this to be the case here - most users won't generally use this functionality, but some with more special or advanced cases will (and some users do have weird and custom inputformats and non-structured data).
The ```Converter``` interface is only really used when the data being read is of some special format (such as HBase and Cassandra binary data). For example, reading Elasticsearch "just works" and uses the standard converter to convert the Writables to primitives. So using ES with SchemaRDD is as easy as below (and this is exactly how it would work in Scala/Java too):
```python
from pyspark.sql import SQLContext
sqlCtx = SQLContext(sc)
conf = {"es.resource" : "index/type"} # assume Elasticsearch is running on localhost defaults
rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat", "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
rdd.first() # the result is a MapWritable that is converted to a Python dict
(u'Elasticsearch ID',
{u'field1': True,
u'field2': u'Some Text',
u'field3': 12345})
dicts = rdd.map(lambda x: x[1])
table = sqlCtx.inferSchema(dicts)
table.registerAsTable("table")
```
@mateiz I agree that the ```Converter``` interface is only an extension point to be used with cases where the inputformat generates data that is not Pickleable (or is e.g. binary data that needs to be extracted), and that we shouldn't add any to Spark itself - the Cassandra and HBase ones are just examples of what can be done.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45357583
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40879001
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14259/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878604
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-43988941
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15163/
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45420334
I personally think Converter is good to keep because it's the only way to expose arbitrary InputFormats to Python. In Java and Scala, you can always call SparkContext.hadoopRDD to get an RDD of Java objects of whatever type there was in the InputFormat. That's why no such interface is needed there. But in Python, there needs to be a way to turn the Java type into a Pickle-able object, and this way seems as good as any.
We should just document this as being an advanced features and less preferred than using Spark SQL when Spark SQL supports the given data source.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41199581
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45574901
Thanks Nick for the latest changes and Patrick for the comments! I've merged this in now. This will be an awesome feature to have.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147713
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
+ """
+ Read a Hadoop SequenceFile with arbitrary key and value Writable class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is as follows:
+ 1. A Java RDD is created from the SequenceFile or other InputFormat, and the key and value Writable classes
+ 2. Serialization is attempted via Pyrolite pickling
+ 3. If this fails, the fallback is to call 'toString' on each key and value
+ 4. C{PickleSerializer} is used to deserialize pickled objects on the Python side
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfint/").collect())
+ [(1, u'aa'), (1, u'aa'), (2, u'aa'), (2, u'bb'), (2, u'bb'), (3, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfdouble/").collect())
+ [(1.0, u'aa'), (1.0, u'aa'), (2.0, u'aa'), (2.0, u'bb'), (2.0, u'bb'), (3.0, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sftext/").collect())
+ [(u'1', u'aa'), (u'1', u'aa'), (u'2', u'aa'), (u'2', u'bb'), (u'2', u'bb'), (u'3', u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfbool/").collect())
+ [(1, False), (1, True), (2, False), (2, False), (2, True), (3, True)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfnull/").collect())
+ [(1, None), (1, None), (2, None), (2, None), (2, None), (3, None)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfmap/").collect())
+ [(1, {2.0: u'aa'}), (1, {3.0: u'bb'}), (2, {1.0: u'aa'}), (2, {1.0: u'cc'}), (2, {3.0: u'bb'}), (3, {2.0: u'dd'})]
+ >>> r = sorted(sc.sequenceFile(tempdir + "/sftestdata/sfclass").collect())[0]
+ >>> r == (u'1', {u'__class__': u'org.apache.spark.api.python.TestWritable', u'double': 54.0, u'int': 123, u'str': u'test1'})
+ True
--- End diff --
Nit: instead of having doctests here, it would be better to put a test in `pyspark/tests.py` for this particular case. The reason is that doctests should ideally be self-contained and should also serve as documentation, but these depend on some files that the reader of the docs won't know about.
Instead, maybe add `@param` statements for the arguments. Look at `pyspark/conf.py` to see how to format these in Epydoc.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148511
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
--- End diff --
Also if we infer the key class and value class for SequenceFiles, should the defaults here be None? The doc and tests imply that we infer them, but we are passing some values here, which is confusing.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45412791
Merged build finished. All automated tests passed.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41147651
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14373/
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41199583
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14379/
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50419281
Not sure if its related, but when I try to start workers having built from trunk, I get this:
[hivedata@hivecluster2 spark]$ ./bin/spark-class org.apache.spark.deploy.worker.Worker spark://hivecluster2:8080
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0
14/07/28 17:05:48 INFO Worker: Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
14/07/28 17:05:48 INFO Worker: Registered signal handlers for [TERM, HUP, INT]
14/07/28 17:05:49 INFO SecurityManager: Changing view acls to: hivedata
14/07/28 17:05:49 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hivedata)
14/07/28 17:05:49 INFO Slf4jLogger: Slf4jLogger started
14/07/28 17:05:49 INFO Remoting: Starting remoting
14/07/28 17:05:49 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkWorker@hivecluster2:36500]
14/07/28 17:05:49 INFO Worker: Starting Spark worker hivecluster2:36500 with 8 cores, 14.6 GB RAM
14/07/28 17:05:49 INFO Worker: Spark home: /home/hivedata/spark
14/07/28 17:05:50 INFO WorkerWebUI: Started WorkerWebUI at http://hivecluster2:8081
14/07/28 17:05:50 INFO Worker: Connecting to master spark://hivecluster2:8080...
14/07/28 17:05:50 ERROR EndpointWriter: AssociationError [akka.tcp://sparkWorker@hivecluster2:36500] -> [akka.tcp://sparkMaster@hivecluster2:8080]: Error [Association failed with [akka.tcp://sparkMaster@hivecluster2:8080]] [
akka.remote.EndpointAssociationException: Association failed with [akka.tcp://sparkMaster@hivecluster2:8080]
Caused by: akka.remote.transport.AkkaProtocolException: The remote system explicitly disassociated (reason unknown).
]
14/07/28 17:05:50 INFO LocalActorRef: Message [akka.remote.transport.AssociationHandle$Disassociated] from Actor[akka://sparkWorker/deadLetters] to Actor[akka://sparkWorker/system/transports/akkaprotocolmanager.tcp0/akkaProtocol-tcp%3A%2F%2FsparkMaster%40hivecluster2%3A8080-1#-396559240] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
14/07/28 17:05:50 ERROR EndpointWriter: AssociationError [akka.tcp://sparkWorker@hivecluster2:36500] -> [akka.tcp://sparkMaster@hivecluster2:8080]: Error [Association failed with [akka.tcp://sparkMaster@hivecluster2:8080]] [
akka.remote.EndpointAssociationException: Association failed with [akka.tcp://sparkMaster@hivecluster2:8080]
Caused by: akka.remote.transport.AkkaProtocolException: The remote system explicitly disassociated (reason unknown).
]
14/07
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41132805
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by ahirreddy <gi...@git.apache.org>.
Github user ahirreddy commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40889685
I haven't had a chance to look too deeply, but I think Pyrolite can be useful. One benefit, and addition to not depending on msg-pack, is that we can do the RDD to PythonRDD conversion in the JVM without calling out to Python. In the SparkSQL API, I basically convert and RDD of java.util.Maps to Python dictionaries. If we can read ```Writable```s into an RDD of java objects, it can easily be converted to a PythonRDD.
The list of supported java-python type mappings is listed here: https://github.com/irmen/Pyrolite/
Here are the java-to-python and vice versa functions I added in #363:
https://github.com/apache/spark/pull/363/files?w=1#diff-0a67bc4d171abe4df8eb305b0f4123a2R290
https://github.com/apache/spark/pull/363/files?w=1#diff-1b97e54687301e5840bb97e576f83ee6R316
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45803692
I had planned to take a look at that next. If you want to take a crack at it I'm happy to review.
You'd have to more or less do things in reverse. The PySpark code for spark SQL does something similar to what would be required, ie takes an RDD of bytes (pickled objects from python) and deserializes them to Java objects. Finally you'll need to convert those to Writables so you can use the relevant output format.
—
Sent from Mailbox
On Wed, Jun 11, 2014 at 1:55 AM, kanzhang <no...@github.com>
wrote:
> @MLnick hey Nick, are you planning on implementing the reverse direction - ```saveAsSequenceFile``` and ```saveAsHadoopFile```? If not, I'd like to give it a shot.
> ---
> Reply to this email directly or view it on GitHub:
> https://github.com/apache/spark/pull/455#issuecomment-45687030
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50425757
Cleaned, made no difference. See https://issues.apache.org/jira/browse/SPARK-1138 where others had this issue.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45432289
As an aside, I am generally -1 on adding a lot of specific reading/writing code to Spark core.
My view is, that is why InputFormat/OutputFormat support is there - to provide that custom read/write functionality. Now it makes sense for something like Parquet with SparkSQL as the preferred format for efficiency (in much the same way as SequenceFiles are often the preferred format in many Hadoop pipelines), but should Spark core contain standardised methods for .saveAsXXXFile for every format? IMO, no - the examples show how to do things with common formats.
I can see why providing contrib modules for reading/writing structured (RDBMS-like) data via common formats for SparkSQL makes sense, as there will probably be one "correct" way of doing this.
But looking at the HBase PR you referenced, I don't see the value of having that live in Spark. And why is it not simply using an ```OutputFormat``` instead of custom config and writing code? (I might be missing something here, but it seems to add complexity and maintenance burden unnecessarily)
---
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50501423
@rjurney I had the same problem with the master branch so I patched the relevant changes from this pull request into the 1.0.0 release, tossed the AvroGenericConverter class into core/src/main/scala/org/apache/spark/api/python, built it and it worked fine.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147744
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
+ """
+ Read a Hadoop SequenceFile with arbitrary key and value Writable class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is as follows:
+ 1. A Java RDD is created from the SequenceFile or other InputFormat, and the key and value Writable classes
+ 2. Serialization is attempted via Pyrolite pickling
+ 3. If this fails, the fallback is to call 'toString' on each key and value
+ 4. C{PickleSerializer} is used to deserialize pickled objects on the Python side
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfint/").collect())
+ [(1, u'aa'), (1, u'aa'), (2, u'aa'), (2, u'bb'), (2, u'bb'), (3, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfdouble/").collect())
+ [(1.0, u'aa'), (1.0, u'aa'), (2.0, u'aa'), (2.0, u'bb'), (2.0, u'bb'), (3.0, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sftext/").collect())
+ [(u'1', u'aa'), (u'1', u'aa'), (u'2', u'aa'), (u'2', u'bb'), (u'2', u'bb'), (u'3', u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfbool/").collect())
+ [(1, False), (1, True), (2, False), (2, False), (2, True), (3, True)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfnull/").collect())
+ [(1, None), (1, None), (2, None), (2, None), (2, None), (3, None)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfmap/").collect())
+ [(1, {2.0: u'aa'}), (1, {3.0: u'bb'}), (2, {1.0: u'aa'}), (2, {1.0: u'cc'}), (2, {3.0: u'bb'}), (3, {2.0: u'dd'})]
+ >>> r = sorted(sc.sequenceFile(tempdir + "/sftestdata/sfclass").collect())[0]
+ >>> r == (u'1', {u'__class__': u'org.apache.spark.api.python.TestWritable', u'double': 54.0, u'int': 123, u'str': u'test1'})
+ True
+ """
+ minSplits = minSplits or min(self.defaultParallelism, 2)
+ jrdd = self._jvm.PythonRDD.sequenceFile(self._jsc, name, key_class, value_class, key_wrapper, value_wrapper,
+ minSplits)
+ return RDD(jrdd, self, PickleSerializer())
+
+ def newAPIHadoopFile(self, name, inputformat_class, key_class="org.apache.hadoop.io.Text",
+ value_class="org.apache.hadoop.io.Text", key_wrapper="toString",
+ value_wrapper="toString", conf={}):
--- End diff --
Why does this always use toString as the wrapper?
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13426207
--- Diff: docs/programming-guide.md ---
@@ -378,9 +378,82 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+## SequenceFile and Hadoop InputFormats
+
+In addition to reading text files, PySpark supports reading [SequenceFile](http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html)
+and any arbitrary [InputFormat](http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/InputFormat.html).
+
+### Writable Support
+
+PySpark SequenceFile support loads an RDD within Java, and pickles the resulting Java objects using
+[Pyrolite](https://github.com/irmen/Pyrolite/). The following Writables are automatically converted:
+
+<table class="table">
+<tr><th>Writable Type</th><th>Scala Type</th><th>Python Type</th></tr>
+<tr><td>Text</td><td>String</td><td>unicode str</td></tr>
+<tr><td>IntWritable</td><td>Int</td><td>int</td></tr>
+<tr><td>FloatWritable</td><td>Float</td><td>float</td></tr>
+<tr><td>DoubleWritable</td><td>Double</td><td>float</td></tr>
+<tr><td>BooleanWritable</td><td>Boolean</td><td>bool</td></tr>
+<tr><td>BytesWritable</td><td>Array[Byte]</td><td>bytearray</td></tr>
+<tr><td>NullWritable</td><td>null</td><td>None</td></tr>
+<tr><td>ArrayWritable</td><td>Array[T]</td><td>list of primitives, or tuple of objects</td></tr>
+<tr><td>MapWritable</td><td>java.util.Map[K, V]</td><td>dict</td></tr>
+<tr><td>Custom Class</td><td>Custom Class conforming to Java Bean conventions</td>
+ <td>dict of public properties (via JavaBean getters and setters) + __class__ for the class type</td></tr>
+</table>
--- End diff --
Good point, can make it more concise.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45344851
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40973845
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148302
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import scala.util.Try
+import org.apache.spark.rdd.RDD
+import org.apache.spark.Logging
+import scala.util.Success
+import scala.util.Failure
+import net.razorvine.pickle.Pickler
+
+/**
+ * Utilities for serialization / deserialization between Python and Java, using MsgPack.
+ * Also contains utilities for converting [[org.apache.hadoop.io.Writable]] ->
+ * Scala objects and primitives
+ */
+private[python] object SerDeUtil extends Logging {
+
+ /**
+ * Convert an RDD of key-value pairs to an RDD of serialized Python objects, that is usable
+ * by PySpark. By default, if serialization fails, toString is called and the string
+ * representation is serialized
+ */
+ def rddToPython[K, V](rdd: RDD[(K, V)]): RDD[Array[Byte]] = {
+ rdd.mapPartitions{ iter =>
+ val pickle = new Pickler
+ var keyFailed = false
+ var valueFailed = false
+ var firstRecord = true
+ iter.map{ case (k, v) =>
+ if (firstRecord) {
+ Try {
+ pickle.dumps(Array(k, v))
+ } match {
+ case Success(b) =>
+ case Failure(err) =>
+ val kt = Try {
+ pickle.dumps(k)
+ }
+ val vt = Try {
+ pickle.dumps(v)
+ }
+ (kt, vt) match {
+ case (Failure(kf), Failure(vf)) =>
+ log.warn(s"""Failed to pickle Java object as key: ${k.getClass.getSimpleName};
--- End diff --
You can use `logWarning` instead.
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13513810
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
+
+ var keyConverter: Converter[Any, Any] = DefaultConverter
+ var valueConverter: Converter[Any, Any] = DefaultConverter
+
+ def convertKey(obj: Any): Any = keyConverter.convert(obj)
+
+ def convertValue(obj: Any): Any = valueConverter.convert(obj)
+
+ def registerKeyConverter(converterClass: String) = {
+ keyConverter = register(converterClass)
+ logInfo(s"Loaded and registered key converter ($converterClass)")
+ }
+
+ def registerValueConverter(converterClass: String) = {
+ valueConverter = register(converterClass)
+ logInfo(s"Loaded and registered value converter ($converterClass)")
+ }
+
+ private def register(converterClass: String): Converter[Any, Any] = {
+ Try {
+ val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
+ converter
+ } match {
+ case Success(s) => s
+ case Failure(err) =>
+ logError(s"Failed to register converter: $converterClass")
+ throw err
+ }
+ }
+}
+
+/** Utilities for working with Python objects -> Hadoop-related objects */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a [[java.util.Map]] of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]): Configuration = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration): Configuration = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)],
+ keyClass: String,
+ keyConverter: Option[String],
+ valueClass: String,
+ valueConverter: Option[String]) = {
+ rdd.mapPartitions { case iter =>
+ val registry = new ConverterRegistry
+ keyConverter.foreach(registry.registerKeyConverter(_))
+ valueConverter.foreach(registry.registerValueConverter(_))
--- End diff --
I'm not sure what value the registry provides above just doing this.
```
def convertRDD[K, V](rdd: RDD[(K, V)],
keyClass: String,
keyConverter: Option[String],
valueClass: String,
valueConverter: Option[String]) = {
iter.map { case (k, v) => (keyConverter.convert(k), valueConverter.convert(v)) }
```
---
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-1416: PySpark support for SequenceFile a...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50515231
@rjurney Yes I doubt in general that patches necessarily magically apply cleanly across versions. You may have to massage it as with any merge conflict.
---
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49699078
@rjurney Nice job adding the DOUBLE and FLOAT. I might be able to get the UNIONS working if you post an example .avro file for me to test it with. I neglected the data structures that I didn't happen to be using.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45420441
(Also, once Spark SQL supports a given interface, or even before, we can simply leave out patches that add new Converters. There's no need to provide a huge collection of them, in fact I'd provide none beyond the default one. The examples here are meant to show how to write your own converter.)
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41213710
Merged build finished. All automated tests passed.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148780
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import scala.util.Try
+import org.apache.spark.rdd.RDD
+import org.apache.spark.Logging
+import scala.util.Success
+import scala.util.Failure
+import net.razorvine.pickle.Pickler
+
+/**
+ * Utilities for serialization / deserialization between Python and Java, using MsgPack.
--- End diff --
This is actually using Pickle now instead of MsgPack
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148747
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkContext
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+
+
+/**
+ * Utilities for working with Python objects -> Hadoop-related objects
+ */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a Map of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]) = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration) = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)]) = {
+ rdd.map{
+ case (k: Writable, v: Writable) => (convert(k).asInstanceOf[K], convert(v).asInstanceOf[V])
+ case (k: Writable, v) => (convert(k).asInstanceOf[K], v.asInstanceOf[V])
+ case (k, v: Writable) => (k.asInstanceOf[K], convert(v).asInstanceOf[V])
+ case (k, v) => (k.asInstanceOf[K], v.asInstanceOf[V])
+ }
+ }
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convert(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convert(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) => (convert(k), convert(v)) }.toMap)
+ case other => other
--- End diff --
It's sort of annoying from a performance point of view that we instantiate these objects for each record as we call `convert`, but we can leave this as is for now and optimize later. I guess the solution would be to look at they key and value of the first record in `convertRDD`, similar to `rddToPython`.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49700443
@ericgarcia Thanks! Very exciting. An example file is here: https://drive.google.com/file/d/0B3wy0wXNwbpRekJVaW13cGRKb1U/edit?usp=sharing
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41144945
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-42790245
Hey Nick, sorry I still haven't looked much at this, been delayed with other 1.0 stuff. I'll get to it when I can though (or get someone else to try it).
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41189552
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-1416: PySpark support for SequenceFile a...
Posted by freedafeng <gi...@git.apache.org>.
Github user freedafeng commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-54686288
Anyone still working on this patch? Pyspark + Hbase is the key to our data science application. I really hope it can work in the very near future.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: SPARK-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46789138
Thanks, master doesn't build for me. Is there a particular commit you
recommend using?
[error]
[error] last tree to typer:
Literal(Constant(org.apache.spark.sql.catalyst.types.PrimitiveType))
[error] symbol: null
[error] symbol definition: null
[error] tpe:
Class(classOf[org.apache.spark.sql.catalyst.types.PrimitiveType])
[error] symbol owners:
[error] context owners: object TestSQLContext -> package test
[error]
[error] == Enclosing template or block ==
[error]
[error] Template( // val <local TestSQLContext>: <notype> in object
TestSQLContext, tree.tpe=org.apache.spark.sql.test.TestSQLContext.type
[error] "org.apache.spark.sql.SQLContext" // parents
[error] ValDef(
[error] private
[error] "_"
[error] <tpt>
[error] <empty>
[error] )
[error] // 2 statements
[error] DefDef( // private def readResolve(): Object in object
TestSQLContext
[error] <method> private <synthetic>
[error] "readResolve"
[error] []
[error] List(Nil)
[error] <tpt> // tree.tpe=Object
[error] test.this."TestSQLContext" // object TestSQLContext in package
test, tree.tpe=org.apache.spark.sql.test.TestSQLContext.type
[error] )
[error] DefDef( // def <init>():
org.apache.spark.sql.test.TestSQLContext.type in object TestSQLContext
[error] <method>
[error] "<init>"
[error] []
[error] List(Nil)
[error] <tpt> // tree.tpe=org.apache.spark.sql.test.TestSQLContext.type
[error] Block( // tree.tpe=Unit
[error] Apply( // def <init>(sparkContext:
org.apache.spark.SparkContext): org.apache.spark.sql.SQLContext in class
SQLContext, tree.tpe=org.apache.spark.sql.SQLContext
[error] TestSQLContext.super."<init>" // def <init>(sparkContext:
org.apache.spark.SparkContext): org.apache.spark.sql.SQLContext in class
SQLContext, tree.tpe=(sparkContext:
org.apache.spark.SparkContext)org.apache.spark.sql.SQLContext
[error] Apply( // def <init>(master: String,appName: String,conf:
org.apache.spark.SparkConf): org.apache.spark.SparkContext in class
SparkContext, tree.tpe=org.apache.spark.SparkContext
[error] new org.apache.spark.SparkContext."<init>" // def
<init>(master: String,appName: String,conf: org.apache.spark.SparkConf):
org.apache.spark.SparkContext in class SparkContext, tree.tpe=(master:
String, appName: String, conf:
org.apache.spark.SparkConf)org.apache.spark.SparkContext
[error] // 3 arguments
[error] "local"
[error] "TestSQLContext"
[error] Apply( // def <init>(): org.apache.spark.SparkConf in
class SparkConf, tree.tpe=org.apache.spark.SparkConf
[error] new org.apache.spark.SparkConf."<init>" // def
<init>(): org.apache.spark.SparkConf in class SparkConf,
tree.tpe=()org.apache.spark.SparkConf
[error] Nil
[error] )
[error] )
[error] )
[error] ()
[error] )
[error] )
[error] )
[error]
[error] == Expanded type of tree ==
[error]
[error] ConstantType(
[error] value =
Constant(org.apache.spark.sql.catalyst.types.PrimitiveType)
[error] )
[error]
[error] uncaught exception during compilation: java.lang.AssertionError
java.lang.AssertionError: assertion failed: List(object package$DebugNode,
object package$DebugNode)
at scala.reflect.internal.Symbols$Symbol.suchThat(Symbols.scala:1678)
at
scala.reflect.internal.Symbols$ClassSymbol.companionModule0(Symbols.scala:2988)
at
scala.reflect.internal.Symbols$ClassSymbol.companionModule(Symbols.scala:2991)
at
scala.tools.nsc.backend.jvm.GenASM$JPlainBuilder.genClass(GenASM.scala:1371)
at scala.tools.nsc.backend.jvm.GenASM$AsmPhase.run(GenASM.scala:120)
at scala.tools.nsc.Global$Run.compileUnitsInternal(Global.scala:1583)
at scala.tools.nsc.Global$Run.compileUnits(Global.scala:1557)
at scala.tools.nsc.Global$Run.compileSources(Global.scala:1553)
at scala.tools.nsc.Global$Run.compile(Global.scala:1662)
at xsbt.CachedCompiler0.run(CompilerInterface.scala:123)
at xsbt.CachedCompiler0.run(CompilerInterface.scala:99)
at xsbt.CompilerInterface.run(CompilerInterface.scala:27)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at sbt.compiler.AnalyzingCompiler.call(AnalyzingCompiler.scala:102)
at sbt.compiler.AnalyzingCompiler.compile(AnalyzingCompiler.scala:48)
at sbt.compiler.AnalyzingCompiler.compile(AnalyzingCompiler.scala:41)
at
sbt.compiler.AggressiveCompile$$anonfun$3$$anonfun$compileScala$1$1.apply$mcV$sp(AggressiveCompile.scala:99)
at
sbt.compiler.AggressiveCompile$$anonfun$3$$anonfun$compileScala$1$1.apply(AggressiveCompile.scala:99)
at
sbt.compiler.AggressiveCompile$$anonfun$3$$anonfun$compileScala$1$1.apply(AggressiveCompile.scala:99)
at
sbt.compiler.AggressiveCompile.sbt$compiler$AggressiveCompile$$timed(AggressiveCompile.scala:166)
at
sbt.compiler.AggressiveCompile$$anonfun$3.compileScala$1(AggressiveCompile.scala:98)
at
sbt.compiler.AggressiveCompile$$anonfun$3.apply(AggressiveCompile.scala:143)
at
sbt.compiler.AggressiveCompile$$anonfun$3.apply(AggressiveCompile.scala:87)
at sbt.inc.IncrementalCompile$$anonfun$doCompile$1.apply(Compile.scala:39)
at sbt.inc.IncrementalCompile$$anonfun$doCompile$1.apply(Compile.scala:37)
at sbt.inc.IncrementalCommon.cycle(Incremental.scala:99)
at sbt.inc.Incremental$$anonfun$1.apply(Incremental.scala:38)
at sbt.inc.Incremental$$anonfun$1.apply(Incremental.scala:37)
at sbt.inc.Incremental$.manageClassfiles(Incremental.scala:65)
at sbt.inc.Incremental$.compile(Incremental.scala:37)
at sbt.inc.IncrementalCompile$.apply(Compile.scala:27)
at sbt.compiler.AggressiveCompile.compile2(AggressiveCompile.scala:157)
at sbt.compiler.AggressiveCompile.compile1(AggressiveCompile.scala:71)
at sbt.compiler.AggressiveCompile.apply(AggressiveCompile.scala:46)
at sbt.Compiler$.apply(Compiler.scala:75)
at sbt.Compiler$.apply(Compiler.scala:66)
at sbt.Defaults$.sbt$Defaults$$compileTaskImpl(Defaults.scala:743)
at sbt.Defaults$$anonfun$compileTask$1.apply(Defaults.scala:735)
at sbt.Defaults$$anonfun$compileTask$1.apply(Defaults.scala:735)
at scala.Function1$$anonfun$compose$1.apply(Function1.scala:47)
at sbt.$tilde$greater$$anonfun$$u2219$1.apply(TypeFunctions.scala:42)
at sbt.std.Transform$$anon$4.work(System.scala:64)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:237)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:237)
at sbt.ErrorHandling$.wideConvert(ErrorHandling.scala:18)
at sbt.Execute.work(Execute.scala:244)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:237)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:237)
at
sbt.ConcurrentRestrictions$$anon$4$$anonfun$1.apply(ConcurrentRestrictions.scala:160)
at sbt.CompletionService$$anon$2.call(CompletionService.scala:30)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:724)
[error] (sql/compile:compile) java.lang.AssertionError: assertion failed:
List(object package$DebugNode, object package$DebugNode)
[error] Total time: 55 s, completed Jun 22, 2014 11:45:36 AM
On Sat, Jun 21, 2014 at 11:17 PM, MLnick <no...@github.com> wrote:
> 1.1 is not released yet. This PR is in master but not in 1.0 (it may be
> released in 1.0.1 or if not then 1.1).
>
>
> So you'll have to clone master and run sbt/sbt publish-local which will
> publish the maven and sbt artifacts to your local repos.
>
>
>
> —
> Sent from Mailbox
>
> On Sun, Jun 22, 2014 at 1:22 AM, Russell Jurney <no...@github.com>
>
> wrote:
>
> > Thanks a ton! One thing - how can I pull spark core 1.1 from maven?
> > [ERROR] Failed to execute goal on project avro: Could not resolve
> > dependencies for project example:avro:jar:0.1: Could not find artifact
> > org.apache.spark:spark-core_2.10:jar:1.1.0-SNAPSHOT in scala-tools.org
> (
> > http://scala-tools.org/repo-releases) -> [Help 1]
> > On Fri, Jun 20, 2014 at 10:45 PM, MLnick <no...@github.com>
> wrote:
> >> @rjurney <https://github.com/rjurney> this works for me (building
> Spark
> >> from current master):
> https://gist.github.com/MLnick/5864741781b9340cb211
> >>
> >> if you run mvn package and then add that to SPARK_CLASSPATH and use it
> in
> >> IPython console.
> >>
> >> However it seems to come through as only strings (not a dict). I
> verified
> >> that if I take only the string field and explicitly convert to string
> (ie Map[String,
> >> String]) then it works. I suspect then that Avro doesn't have the type
> >> information at all, so Pyrolite cannot pickle it. I guess you might
> have to
> >> do something more in depth in the AvroConverter to read the type info
> >> from the Avro schema and do a cast...
> >>
> >> —
> >> Reply to this email directly or view it on GitHub
> >> <https://github.com/apache/spark/pull/455#issuecomment-46745394>.
> >>
> > --
> > Russell Jurney twitter.com/rjurney russell.jurney@gmail.com
> datasyndrome.com
> > ---
> > Reply to this email directly or view it on GitHub:
> > https://github.com/apache/spark/pull/455#issuecomment-46767642
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/455#issuecomment-46773577>.
>
--
Russell Jurney twitter.com/rjurney russell.jurney@gmail.com datasyndrome.com
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46638552
It would be really helpful if this enabled loading of Avro data via GenericRecord.
---
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-1416: PySpark support for SequenceFile a...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50422265
You're not literally writing 4.X.X in the version are you?
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40937437
@mateiz the binary files are generated with the ```WriteInputFormatTestDataGenerator``` main method. But they are generated once and saved.
Do you prefer to generate them as a setup before test run, and tear them down afterward?
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41204035
This error in the latest Jenkins run:
```
File "pyspark/context.py", line 293, in __main__.SparkContext.wholeTextFiles
Failed example:
sorted(textFiles.collect())
Expected:
[(u'.../1.txt', u'1'), (u'.../2.txt', u'2')]
Got:
[(u'/tmp/tmps6Hn0G/files/1.txt', u'1'), (u'/tmp/tmps6Hn0G/files/2.txt', u'2')]
```
Means that the "ellipsis" option of DocTest isn't set -- not sure why, because it was set before and this was passing. But maybe it got triggered by some earlier errors somehow.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415318
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
--- End diff --
Also, add a ScalaDoc to this and add `:: Experimental ::` at the beginning of the doc and the annotation `@ExperimentalAPI`. Then it will be marked experimental in the docs, similar to https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.FutureAction.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41203951
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13521493
--- Diff: docs/programming-guide.md ---
@@ -378,11 +378,88 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
-</div>
+### SequenceFile and Hadoop InputFormats
--- End diff --
Further up in this guide there is a statement that says:
```
The current API is limited to text files, but support for binary Hadoop InputFormats is expected in future versions.
```
Given this patch, it probably makes sense to remove 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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13414789
--- Diff: docs/programming-guide.md ---
@@ -378,9 +378,82 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+## SequenceFile and Hadoop InputFormats
--- End diff --
You probably need to use `###` for subheadings since the one above was already `##`; see how it looks in the doc once built (you can do `SKIP_API=1 jekyll serve --watch` in `docs/` and then go to http://localhost:4000).
---
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50541143
@rjurney For some reason I can't push my changes to a tagged release in github. Here's the packaged distribution that I built: https://dl.dropboxusercontent.com/u/145006/spark-1.0.0-bin-1.0.4.tgz it's built for hadoop 1.0
If you want to create your own modified 1.0.0 release, what i did was: check out the tagged 1.0.0 release https://github.com/apache/spark/tree/v1.0.0 and also check out this pull request and then copy the files from this pull request into the corresponding directories in the 1.0.0 release. Then create `org.apache.spark.api.python.AvroGenericConverter` in the same directory as the other `org.apache.spark.api.python` namespace files, change the namespace on `AvroGenericConverter` in the code accordingly, and build the project with `./make_distribution.sh`
The modified code for loading the avro files in pyspark is
avroRdd = sc.newAPIHadoopFile(filename,
"org.apache.avro.mapreduce.AvroKeyInputFormat",
"org.apache.avro.mapred.AvroKey",
"org.apache.hadoop.io.NullWritable",
keyConverter="org.apache.spark.api.python.AvroGenericConverter")
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46656191
@rjurney it should work but you'll need to write a converter like:
```scala
class AvroConverter extends Converter[AvroKey[GenericRecord], java.util.Map[String, Any]] {
import collection.JavaConversions._
override def convert(obj: AvroKey[GenericRecord]): java.util.Map[String, Any] = {
val record = obj.datum()
val schema = record.getSchema
mapAsJavaMap(schema.getFields.map( f => (f.name, record.get(f.name)) ).toMap)
}
}
```
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13518669
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
+
+ var keyConverter: Converter[Any, Any] = DefaultConverter
+ var valueConverter: Converter[Any, Any] = DefaultConverter
+
+ def convertKey(obj: Any): Any = keyConverter.convert(obj)
+
+ def convertValue(obj: Any): Any = valueConverter.convert(obj)
+
+ def registerKeyConverter(converterClass: String) = {
+ keyConverter = register(converterClass)
+ logInfo(s"Loaded and registered key converter ($converterClass)")
+ }
+
+ def registerValueConverter(converterClass: String) = {
+ valueConverter = register(converterClass)
+ logInfo(s"Loaded and registered value converter ($converterClass)")
+ }
+
+ private def register(converterClass: String): Converter[Any, Any] = {
+ Try {
+ val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
+ converter
+ } match {
+ case Success(s) => s
+ case Failure(err) =>
+ logError(s"Failed to register converter: $converterClass")
+ throw err
+ }
+ }
+}
+
+/** Utilities for working with Python objects -> Hadoop-related objects */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a [[java.util.Map]] of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]): Configuration = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration): Configuration = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)],
+ keyClass: String,
+ keyConverter: Option[String],
+ valueClass: String,
+ valueConverter: Option[String]) = {
+ rdd.mapPartitions { case iter =>
+ val registry = new ConverterRegistry
+ keyConverter.foreach(registry.registerKeyConverter(_))
+ valueConverter.foreach(registry.registerValueConverter(_))
--- End diff --
Yes, the registry idea originally came from the fact that a new ```WritableConverter``` was being instantiated on each call to ```convert``` (see L48). I was trying to make it so that one converter was instantiated and cached in the registry for reuse.
I haven't gotten to that part yet but I was thinking of getting this in and improving that efficiency later on.
I'll take a look at using ```RDD.first``` rather to inspect the types and instantiate the correct ```WritableConverter``` in the case of the ```DefaultConverter```
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13514748
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import scala.util.Try
+import org.apache.spark.rdd.RDD
+import org.apache.spark.Logging
+import scala.util.Success
+import scala.util.Failure
+import net.razorvine.pickle.Pickler
+
+
+/** Utilities for serialization / deserialization between Python and Java, using Pickle. */
+private[python] object SerDeUtil extends Logging {
+
+ /**
+ * Convert an RDD of key-value pairs to an RDD of serialized Python objects, that is usable
+ * by PySpark. By default, if serialization fails, toString is called and the string
+ * representation is serialized
+ */
+ def rddToPython[K, V](rdd: RDD[(K, V)]): RDD[Array[Byte]] = {
+ rdd.mapPartitions { iter =>
--- End diff --
Would it be possible to, instead of trying to pickle the first object of each partition, instead just try to take the first element in the RDD and pickle it's key and value. This would be cleaner and also avoid some potentially weird failure conditions if, e.g. for some reason we make different decisions for different partitions. If you check out the SQLContext this is exactly what it does:
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala#L299
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45160411
Hey Nick, this looks really great! Thanks for taking the time to write examples too. I made some comments on the Converter API and docs but I think it will be more or less ready to go soon.
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45414134
Hey @MLnick thanks for making those changes. This is looking good! I didn't get to totally finish the review yesterday but I'll make sure I finish it today
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50422374
@srowen Actually yes, I'm that stupid :) Figured it out on me own though, have it building across the cluster 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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r11919136
--- Diff: python/pyspark/context.py ---
@@ -467,8 +555,9 @@ def _test():
globs = globals().copy()
globs['sc'] = SparkContext('local[4]', 'PythonTest', batchSize=2)
globs['tempdir'] = tempfile.mkdtemp()
+ globs['sc']._jvm.WriteInputFormatTestDataGenerator.generateData(globs['tempdir'], globs['sc']._jsc)
atexit.register(lambda: shutil.rmtree(globs['tempdir']))
- (failure_count, test_count) = doctest.testmod(globs=globs, optionflags=doctest.ELLIPSIS)
+ (failure_count, test_count) = doctest.testmod(globs=globs)
--- End diff --
Ah, actually you *did* remove the ellipsis option, right here! You need to add it back for other tests to work.
BTW you can run just the Python tests locally by doing sbt assembly/assembly and then python/run-tests. You can also run pyspark context.py to run just the tests in that file. It's much faster than running through the whole test suite.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50422022
@JoshRosen Actually, I just did: sbt/sbt assembly publish-local
Trying again with: SPARK_HADOOP_VERSION=2.0.0-mr1-cdh4.X.X sbt/sbt assembly publish-local
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13520371
--- Diff: core/src/main/scala/org/apache/spark/api/python/WriteInputFormatTestDataGenerator.scala ---
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.SparkContext
+import org.apache.hadoop.io._
+import scala.Array
+import java.io.{DataOutput, DataInput}
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
+import org.apache.spark.api.java.JavaSparkContext
+
+/**
+ * A class to test MsgPack serialization on the Scala side, that will be deserialized
+ * in Python
+ * @param str
+ * @param int
+ * @param double
+ */
+case class TestWritable(var str: String, var int: Int, var double: Double) extends Writable {
--- End diff --
All the classes in this file should be `private[spark]` since they are not in the test package.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41136745
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14364/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41010120
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44487284
By the way one other change: the parameter names should use `camelCase` to be consistent with our other ones. In PySpark we just chose to keep the same naming style as in Scala / Java to make code easier to translate between the languages.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40973977
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49687447
@rjurney I saved the above to
`spark/examples/src/main/scala/org/apache/spark/examples/pythonconverters/AvroGenericConverter.scala`
rebuilt spark with
`./sbt/sbt assembly`
and then loaded pyspark with
`SPARK_CLASSPATH=examples/target/scala-2.10/spark-examples-1.1.0-SNAPSHOT-hadoop1.0.4.jar IPYTHON=1 bin/pyspark`
Inside of pyspark you should be able to load up yourAvro file with:
avroRdd = sc.newAPIHadoopFile(<full_path_to_your_avro_file>,
"org.apache.avro.mapreduce.AvroKeyInputFormat",
"org.apache.avro.mapred.AvroKey",
"org.apache.hadoop.io.NullWritable",
keyConverter="org.apache.spark.examples.pythonconverters.AvroGenericConverter")
Let me know if it works for you : )
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148122
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkContext
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+
+
+/**
+ * Utilities for working with Python objects -> Hadoop-related objects
+ */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a Map of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]) = {
--- End diff --
We've updated our style guide to require specifying return types on any function, so you should add them on these. Or use `: Unit` if they return `Unit`. The reason to do it is that it makes it easier to ensure binary compatibility (otherwise you can end up with some weird types and changes to them when you change the body of the method).
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13426202
--- Diff: docs/programming-guide.md ---
@@ -378,9 +378,82 @@ Some notes on reading files with Spark:
* The `textFile` method also takes an optional second argument for controlling the number of slices of the file. By default, Spark creates one slice for each block of the file (blocks being 64MB by default in HDFS), but you can also ask for a higher number of slices by passing a larger value. Note that you cannot have fewer slices than blocks.
-Apart reading files as a collection of lines,
+Apart from reading files as a collection of lines,
`SparkContext.wholeTextFiles` lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with `textFile`, which would return one record per line in each file.
+## SequenceFile and Hadoop InputFormats
+
+In addition to reading text files, PySpark supports reading [SequenceFile](http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html)
+and any arbitrary [InputFormat](http://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/InputFormat.html).
+
+### Writable Support
+
+PySpark SequenceFile support loads an RDD within Java, and pickles the resulting Java objects using
+[Pyrolite](https://github.com/irmen/Pyrolite/). The following Writables are automatically converted:
+
+<table class="table">
+<tr><th>Writable Type</th><th>Scala Type</th><th>Python Type</th></tr>
+<tr><td>Text</td><td>String</td><td>unicode str</td></tr>
+<tr><td>IntWritable</td><td>Int</td><td>int</td></tr>
+<tr><td>FloatWritable</td><td>Float</td><td>float</td></tr>
+<tr><td>DoubleWritable</td><td>Double</td><td>float</td></tr>
+<tr><td>BooleanWritable</td><td>Boolean</td><td>bool</td></tr>
+<tr><td>BytesWritable</td><td>Array[Byte]</td><td>bytearray</td></tr>
+<tr><td>NullWritable</td><td>null</td><td>None</td></tr>
+<tr><td>ArrayWritable</td><td>Array[T]</td><td>list of primitives, or tuple of objects</td></tr>
+<tr><td>MapWritable</td><td>java.util.Map[K, V]</td><td>dict</td></tr>
+<tr><td>Custom Class</td><td>Custom Class conforming to Java Bean conventions</td>
+ <td>dict of public properties (via JavaBean getters and setters) + __class__ for the class type</td></tr>
+</table>
+
+### Loading SequenceFiles
+
+Similarly to text files, SequenceFiles can be loaded by specifying the path. The key and value
+classes can be specified, but for standard Writables it should work without requiring this.
+
+{% highlight python %}
+>>> rdd = sc.sequenceFile("path/to/sequencefile/of/doubles")
+>>> rdd.collect() # this example has DoubleWritable keys and Text values
+[(1.0, u'aa'),
+ (2.0, u'bb'),
+ (2.0, u'aa'),
+ (3.0, u'cc'),
+ (2.0, u'bb'),
+ (1.0, u'aa')]
+>>> help(sc.sequenceFile) # Show sequencefile documentation
+{% endhighlight %}
+
+### Loading Arbitrary Hadoop InputFormats
+
+PySpark can also read any Hadoop InputFormat, for both 'new' and 'old' Hadoop APIs. If required,
+a Hadoop configuration can be passed in as a Python dict. Here is an example using the
+Elasticsearch ESInputFormat:
+
+{% highlight python %}
+$ SPARK_CLASSPATH=/path/to/elasticsearch-hadoop.jar ./bin/pyspark
+>>> conf = {"es.resource" : "index/type"} # assume Elasticsearch is running on localhost defaults
+>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",\
+ "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedMapWritable", conf=conf)
+>>> rdd.first() # the result is a MapWritable that is converted to a Python dict
+(u'Elasticsearch ID',
+ {u'field1': True,
+ u'field2': u'Some Text',
+ u'field3': 12345})
+>>> help(sc.newAPIHadoopRDD) # Show help for new API Hadoop RDD
+{% endhighlight %}
+
+Note that, if the InputFormat simply depends on a Hadoop configuration and/or input path, and
+the key and value classes can easily be converted according to the above table,
+then this approach should work well for such cases.
+
+If you have custom serialized binary data (like pulling data from Cassandra / HBase) or custom
+classes that don't conform to the JavaBean requirements, then you will probably have to first
+transform that data on the Scala/Java side to something which can be handled by Pyrolite's pickler.
+
+Future support for custom 'converter' functions for keys/values that allows this to be written in Java/Scala,
+and called from Python, as well as support for writing data out as SequenceFileOutputFormat
+and other OutputFormats, is forthcoming.
--- End diff --
Ah sorry missed this. Will amend.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49698499
This file, without any UNIONS, works: https://github.com/miguno/avro-cli-examples/blob/master/twitter.snappy.avro
My data is more complex :(
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49697592
My data has doubles in it, could that be the issue?
Using Python version 2.7.6rc1 (v2.7.6rc1:4913d0e9be30+, Oct 27 2013 20:52:11)
SparkContext available as sc.
In [1]: avroRdd = sc.newAPIHadoopFile('/tmp/part-m-00000.avro',
...: "org.apache.avro.mapreduce.AvroKeyInputFormat",
...: "org.apache.avro.mapred.AvroKey",
...: "org.apache.hadoop.io.NullWritable",
...: keyConverter="org.apache.spark.examples.pythonconverters.AvroGenericConverter")
14/07/21 21:38:48 INFO MemoryStore: ensureFreeSpace(56082) called with curMem=0, maxMem=309225062
14/07/21 21:38:48 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 54.8 KB, free 294.8 MB)
14/07/21 21:38:48 INFO Converter: Loaded converter: org.apache.spark.examples.pythonconverters.AvroGenericConverter
14/07/21 21:38:48 INFO FileInputFormat: Total input paths to process : 1
14/07/21 21:38:48 INFO SparkContext: Starting job: first at SerDeUtil.scala:69
14/07/21 21:38:48 INFO DAGScheduler: Got job 0 (first at SerDeUtil.scala:69) with 1 output partitions (allowLocal=true)
14/07/21 21:38:48 INFO DAGScheduler: Final stage: Stage 0(first at SerDeUtil.scala:69)
14/07/21 21:38:48 INFO DAGScheduler: Parents of final stage: List()
14/07/21 21:38:48 INFO DAGScheduler: Missing parents: List()
14/07/21 21:38:48 INFO DAGScheduler: Computing the requested partition locally
14/07/21 21:38:48 INFO NewHadoopRDD: Input split: file:/tmp/part-m-00000.avro:0+1996482
14/07/21 21:38:48 WARN AvroKeyInputFormat: Reader schema was not set. Use AvroJob.setInputKeySchema() if desired.
14/07/21 21:38:48 INFO AvroKeyInputFormat: Using a reader schema equal to the writer schema.
14/07/21 21:38:48 INFO DAGScheduler: Failed to run first at SerDeUtil.scala:69
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-1-6b7a8c841ffd> in <module>()
3 "org.apache.avro.mapred.AvroKey",
4 "org.apache.hadoop.io.NullWritable",
----> 5 keyConverter="org.apache.spark.examples.pythonconverters.AvroGenericConverter")
/Users/rjurney/Software/spark-trunk/python/pyspark/context.pyc in newAPIHadoopFile(self, path, inputFormatClass, keyClass, valueClass, keyConverter, valueConverter, conf)
402 jconf = self._dictToJavaMap(conf)
403 jrdd = self._jvm.PythonRDD.newAPIHadoopFile(self._jsc, path, inputFormatClass, keyClass,
--> 404 valueClass, keyConverter, valueConverter, jconf)
405 return RDD(jrdd, self, PickleSerializer())
406
/Users/rjurney/Software/spark-trunk/python/lib/py4j-0.8.1-src.zip/py4j/java_gateway.py in __call__(self, *args)
535 answer = self.gateway_client.send_command(command)
536 return_value = get_return_value(answer, self.gateway_client,
--> 537 self.target_id, self.name)
538
539 for temp_arg in temp_args:
/Users/rjurney/Software/spark-trunk/python/lib/py4j-0.8.1-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
298 raise Py4JJavaError(
299 'An error occurred while calling {0}{1}{2}.\n'.
--> 300 format(target_id, '.', name), value)
301 else:
302 raise Py4JError(
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopFile.
: org.apache.spark.SparkDriverExecutionException: Execution error
at org.apache.spark.scheduler.DAGScheduler.runLocallyWithinThread(DAGScheduler.scala:585)
at org.apache.spark.scheduler.DAGScheduler$$anon$1.run(DAGScheduler.scala:563)
Caused by: java.lang.NullPointerException
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.org$apache$spark$examples$pythonconverters$AvroGenericConverter$$unpack$1(AvroGenericConverter.scala:31)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter$$anonfun$unpackRecord$1$1.apply(AvroGenericConverter.scala:17)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter$$anonfun$unpackRecord$1$1.apply(AvroGenericConverter.scala:17)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.AbstractTraversable.map(Traversable.scala:105)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.unpackRecord$1(AvroGenericConverter.scala:17)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.convert(AvroGenericConverter.scala:34)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.convert(AvroGenericConverter.scala:12)
at org.apache.spark.api.python.PythonHadoopUtil$$anonfun$convertRDD$1.apply(PythonHadoopUtil.scala:126)
at org.apache.spark.api.python.PythonHadoopUtil$$anonfun$convertRDD$1.apply(PythonHadoopUtil.scala:126)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1068)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1068)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1096)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1096)
at org.apache.spark.scheduler.DAGScheduler.runLocallyWithinThread(DAGScheduler.scala:578)
... 1 more
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41987966
Sorry, haven't had a chance yet, but will do soon.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41012536
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-51202880
@ericgarcia @srowen @MLnick
Unfortunately when I follow those directions, I still get errors. It looks like I'll have to wait to get this functionality until its merged. Anyone have any idea when that will be? Is there a release in mind that this would be a part of?
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: SPARK-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13514379
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
+
+ var keyConverter: Converter[Any, Any] = DefaultConverter
+ var valueConverter: Converter[Any, Any] = DefaultConverter
+
+ def convertKey(obj: Any): Any = keyConverter.convert(obj)
+
+ def convertValue(obj: Any): Any = valueConverter.convert(obj)
+
+ def registerKeyConverter(converterClass: String) = {
+ keyConverter = register(converterClass)
+ logInfo(s"Loaded and registered key converter ($converterClass)")
+ }
+
+ def registerValueConverter(converterClass: String) = {
+ valueConverter = register(converterClass)
+ logInfo(s"Loaded and registered value converter ($converterClass)")
+ }
+
+ private def register(converterClass: String): Converter[Any, Any] = {
+ Try {
+ val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
+ converter
+ } match {
+ case Success(s) => s
+ case Failure(err) =>
+ logError(s"Failed to register converter: $converterClass")
+ throw err
+ }
+ }
+}
+
+/** Utilities for working with Python objects -> Hadoop-related objects */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a [[java.util.Map]] of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]): Configuration = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration): Configuration = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)],
--- End diff --
I think this should return an `RDD[Any, Any]` not an `RDD[K, V]`. Once we've converted it we've lost the type information. In the code you've added all we do is pass the entries to the picker which just inspects the runtime type, but if anyone tried to do something that actually relied on this returning K or V it would throw a runtime exception.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44743612
@mateiz I noticed that the docs for Python programming guid have been merged into the overall Programming guide.
Where do you think the best place is to put the bit of documentation about InputFormats for PySpark?
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147714
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
+ """
+ Read a Hadoop SequenceFile with arbitrary key and value Writable class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is as follows:
+ 1. A Java RDD is created from the SequenceFile or other InputFormat, and the key and value Writable classes
+ 2. Serialization is attempted via Pyrolite pickling
+ 3. If this fails, the fallback is to call 'toString' on each key and value
+ 4. C{PickleSerializer} is used to deserialize pickled objects on the Python side
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfint/").collect())
+ [(1, u'aa'), (1, u'aa'), (2, u'aa'), (2, u'bb'), (2, u'bb'), (3, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfdouble/").collect())
+ [(1.0, u'aa'), (1.0, u'aa'), (2.0, u'aa'), (2.0, u'bb'), (2.0, u'bb'), (3.0, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sftext/").collect())
+ [(u'1', u'aa'), (u'1', u'aa'), (u'2', u'aa'), (u'2', u'bb'), (u'2', u'bb'), (u'3', u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfbool/").collect())
+ [(1, False), (1, True), (2, False), (2, False), (2, True), (3, True)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfnull/").collect())
+ [(1, None), (1, None), (2, None), (2, None), (2, None), (3, None)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfmap/").collect())
+ [(1, {2.0: u'aa'}), (1, {3.0: u'bb'}), (2, {1.0: u'aa'}), (2, {1.0: u'cc'}), (2, {3.0: u'bb'}), (3, {2.0: u'dd'})]
+ >>> r = sorted(sc.sequenceFile(tempdir + "/sftestdata/sfclass").collect())[0]
+ >>> r == (u'1', {u'__class__': u'org.apache.spark.api.python.TestWritable', u'double': 54.0, u'int': 123, u'str': u'test1'})
+ True
--- End diff --
Nit: instead of having doctests here, it would be better to put a test in `pyspark/tests.py` for this particular case. The reason is that doctests should ideally be self-contained and should also serve as documentation, but these depend on some files that the reader of the docs won't know about.
Instead, maybe add `@param` statements for the arguments. Look at `pyspark/conf.py` to see how to format these in Epydoc.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41145116
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49716773
@ericgarcia hey that's cool. Would you mind creating a pull request to add that to the Spark examples? It would be very useful to others to see how to make a fully generic Avro converter.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415338
--- Diff: examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala ---
@@ -32,6 +32,23 @@ import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
+import org.apache.spark.api.python.Converter
+
+class CassandraCQLKeyConverter extends Converter {
+ import collection.JavaConversions._
--- End diff --
Could probably change this to only import `mapAsJavaMap`.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45099078
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15446/
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13513759
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
--- End diff --
Could this code be simplified by removing this class? Since this is already using a `mapPartitions` block there will only be one instance of each converter created for each partition anyways.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41209690
@mateiz ah something got messed up with all the merges along the way. Fixed that up and PySpark tests all passing locally
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-43985471
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40955147
Yeah, it would be good to generate them during the test and tear them down after (you can just call the Scala class that generates them from Python). It's kind of confusing to include them in git.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41010129
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50084602
When I load my records, I get:
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopFile.
: org.apache.spark.SparkDriverExecutionException: Execution error
at org.apache.spark.scheduler.DAGScheduler.runLocallyWithinThread(DAGScheduler.scala:585)
at org.apache.spark.scheduler.DAGScheduler$$anon$1.run(DAGScheduler.scala:563)
Caused by: scala.MatchError: UNION (of class org.apache.avro.Schema$Type)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.org$apache$spark$examples$pythonconverters$AvroGenericConverter$$unpack$1(AvroGenericConverter.scala:24)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter$$anonfun$unpackRecord$1$1.apply(AvroGenericConverter.scala:17)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter$$anonfun$unpackRecord$1$1.apply(AvroGenericConverter.scala:17)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.AbstractTraversable.map(Traversable.scala:105)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.unpackRecord$1(AvroGenericConverter.scala:17)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.convert(AvroGenericConverter.scala:36)
at org.apache.spark.examples.pythonconverters.AvroGenericConverter.convert(AvroGenericConverter.scala:12)
at org.apache.spark.api.python.PythonHadoopUtil$$anonfun$convertRDD$1.apply(PythonHadoopUtil.scala:126)
at org.apache.spark.api.python.PythonHadoopUtil$$anonfun$convertRDD$1.apply(PythonHadoopUtil.scala:126)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1068)
at org.apache.spark.rdd.RDD$$anonfun$28.apply(RDD.scala:1068)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1096)
at org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1096)
at org.apache.spark.scheduler.DAGScheduler.runLocallyWithinThread(DAGScheduler.scala:578)
... 1 more
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147767
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
+ """
+ Read a Hadoop SequenceFile with arbitrary key and value Writable class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is as follows:
+ 1. A Java RDD is created from the SequenceFile or other InputFormat, and the key and value Writable classes
+ 2. Serialization is attempted via Pyrolite pickling
+ 3. If this fails, the fallback is to call 'toString' on each key and value
+ 4. C{PickleSerializer} is used to deserialize pickled objects on the Python side
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfint/").collect())
+ [(1, u'aa'), (1, u'aa'), (2, u'aa'), (2, u'bb'), (2, u'bb'), (3, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfdouble/").collect())
+ [(1.0, u'aa'), (1.0, u'aa'), (2.0, u'aa'), (2.0, u'bb'), (2.0, u'bb'), (3.0, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sftext/").collect())
+ [(u'1', u'aa'), (u'1', u'aa'), (u'2', u'aa'), (u'2', u'bb'), (u'2', u'bb'), (u'3', u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfbool/").collect())
+ [(1, False), (1, True), (2, False), (2, False), (2, True), (3, True)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfnull/").collect())
+ [(1, None), (1, None), (2, None), (2, None), (2, None), (3, None)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfmap/").collect())
+ [(1, {2.0: u'aa'}), (1, {3.0: u'bb'}), (2, {1.0: u'aa'}), (2, {1.0: u'cc'}), (2, {3.0: u'bb'}), (3, {2.0: u'dd'})]
+ >>> r = sorted(sc.sequenceFile(tempdir + "/sftestdata/sfclass").collect())[0]
+ >>> r == (u'1', {u'__class__': u'org.apache.spark.api.python.TestWritable', u'double': 54.0, u'int': 123, u'str': u'test1'})
+ True
+ """
+ minSplits = minSplits or min(self.defaultParallelism, 2)
+ jrdd = self._jvm.PythonRDD.sequenceFile(self._jsc, name, key_class, value_class, key_wrapper, value_wrapper,
+ minSplits)
+ return RDD(jrdd, self, PickleSerializer())
+
+ def newAPIHadoopFile(self, name, inputformat_class, key_class="org.apache.hadoop.io.Text",
+ value_class="org.apache.hadoop.io.Text", key_wrapper="toString",
+ value_wrapper="toString", conf={}):
--- End diff --
(It seems that newAPIHadoopRDD doesn't for example)
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45101262
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15447/
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45344852
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15508/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40939969
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-1416: PySpark support for SequenceFile a...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-54686436
@freedafeng This PR was actually merged and will be available in Spark 1.1 (which should be released _very_ soon). https://github.com/apache/spark/commit/f971d6cb60d642178d6544217a25fa16ece34889
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: SPARK-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415213
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
--- End diff --
IMO this should be parametrized as `trait Converter[T, U]` and have `def convert(obj: T): U`. This will make it easier to write subclasses in Java or Scala and it won't affect anything below because of type erasure (in the worst case it means a bit more ugly code to pass this stuff through, e.g. casting the converter to a `Converter[Any, Any]`).
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41012538
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14322/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41147649
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41147925
@mateiz I fixed the PySpark tests I believe, but seems this build is
failing for some other reason.
By the way, I guess this will be too late for 1.0.0 but would be great to
get it in the next release if everyone is happy
On Wed, Apr 23, 2014 at 12:50 PM, UCB AMPLab <no...@github.com>wrote:
> Merged build finished.
>
> —
> Reply to this email directly or view it on GitHub<https://github.com/apache/spark/pull/455#issuecomment-41147649>
> .
>
---
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13514061
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+import org.apache.spark.annotation.Experimental
+
+
+/**
+ * :: Experimental ::
+ * A trait for use with reading custom classes in PySpark. Implement this trait and add custom
+ * transformation code by overriding the convert method.
+ */
+@Experimental
+trait Converter[T, U] {
+ def convert(obj: T): U
+}
+
+/**
+ * A converter that handles conversion of common [[org.apache.hadoop.io.Writable]] objects.
+ * Other objects are passed through without conversion.
+ */
+private[python] object DefaultConverter extends Converter[Any, Any] {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+/**
+ * The converter registry holds a key and value converter, so that they are only instantiated
+ * once per RDD partition.
+ */
+private[python] class ConverterRegistry extends Logging {
+
+ var keyConverter: Converter[Any, Any] = DefaultConverter
+ var valueConverter: Converter[Any, Any] = DefaultConverter
+
+ def convertKey(obj: Any): Any = keyConverter.convert(obj)
+
+ def convertValue(obj: Any): Any = valueConverter.convert(obj)
+
+ def registerKeyConverter(converterClass: String) = {
+ keyConverter = register(converterClass)
+ logInfo(s"Loaded and registered key converter ($converterClass)")
+ }
+
+ def registerValueConverter(converterClass: String) = {
+ valueConverter = register(converterClass)
+ logInfo(s"Loaded and registered value converter ($converterClass)")
+ }
+
+ private def register(converterClass: String): Converter[Any, Any] = {
+ Try {
+ val converter = Class.forName(converterClass).newInstance().asInstanceOf[Converter[Any, Any]]
+ converter
+ } match {
+ case Success(s) => s
+ case Failure(err) =>
+ logError(s"Failed to register converter: $converterClass")
+ throw err
+ }
+ }
+}
+
+/** Utilities for working with Python objects -> Hadoop-related objects */
+private[python] object PythonHadoopUtil {
+
+ /**
+ * Convert a [[java.util.Map]] of properties to a [[org.apache.hadoop.conf.Configuration]]
+ */
+ def mapToConf(map: java.util.Map[String, String]): Configuration = {
+ import collection.JavaConversions._
+ val conf = new Configuration()
+ map.foreach{ case (k, v) => conf.set(k, v) }
+ conf
+ }
+
+ /**
+ * Merges two configurations, returns a copy of left with keys from right overwriting
+ * any matching keys in left
+ */
+ def mergeConfs(left: Configuration, right: Configuration): Configuration = {
+ import collection.JavaConversions._
+ val copy = new Configuration(left)
+ right.iterator().foreach(entry => copy.set(entry.getKey, entry.getValue))
+ copy
+ }
+
+ /**
+ * Converts an RDD of key-value pairs, where key and/or value could be instances of
+ * [[org.apache.hadoop.io.Writable]], into an RDD[(K, V)]
+ */
+ def convertRDD[K, V](rdd: RDD[(K, V)],
+ keyClass: String,
+ keyConverter: Option[String],
--- End diff --
Could this just pass a converter object directly instead of a string? As it stands now it's pretty hard to track the types of objects being passed around... why not just add properly typed converters 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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41204451
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44519603
Hi @mateiz
Thanks for taking a look and the comments. I'll make the necessary changes.
On the inference of key/value classes, I will take a deeper look and see what the best approach may be for non-SF formats. Tests with incorrect classes is a good idea to stress test it and see where it breaks.
I had a similar idea in mind for the ```PythonConverter``` interface. I can try to flesh that out a bit for this PR, but I sort of wanted to get something in that is clearly marked as VERY experimental so that we could perhaps get some feedback from usage "in the wild". I suspect there will be many edge cases with this.
Thoughts?
As per the mailing list discussion on HBase, I will try to get an HBase example working to iron out some kinks (http://apache-spark-user-list.1001560.n3.nabble.com/Python-Spark-and-HBase-td6142.html#a6507).
At a minimum, I'd like to have this work relatively easily with ES (already does), HBase, Cassandra.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13414996
--- Diff: examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala ---
@@ -32,6 +32,23 @@ import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
+import org.apache.spark.api.python.Converter
+
+class CassandraCQLKeyConverter extends Converter {
+ import collection.JavaConversions._
+ override def convert(obj: Any) = {
+ val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]]
+ mapAsJavaMap(result.mapValues(bb => ByteBufferUtil.toInt(bb)))
+ }
+}
+
+class CassandraCQLValueConverter extends Converter {
+ import collection.JavaConversions._
+ override def convert(obj: Any) = {
+ val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]]
+ mapAsJavaMap(result.mapValues(bb => ByteBufferUtil.string(bb)))
+ }
+}
--- End diff --
Move these to separate source files since they're not used in the Scala example. Maybe we can even have a "pythonConverters" subpackage of org.apache.spark.examples that says these are used in the .py examples.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-43985454
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45094783
@mateiz thanks for all the comments. I've addressed all of them (I think!).
I moved tests to ```tests.py``` and added a few more tests.
I've also now added the ```Converter``` interface, together with a test.
I used this interface to add an example to the ```examples``` project for reading from HBase ```TableInputFormat``` and Cassandra ```CqlPagingInputFormat```, combined with appropriate ```Converter```s to convert the results to something useful.
Please take a look over things and let me know any further issues you pick up, or 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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50111706
You'd need to run loading code off master branch. It should be in 1.1 release in a few weeks—
Sent from Mailbox
On Fri, Jul 25, 2014 at 4:14 AM, Russell Jurney <no...@github.com>
wrote:
> I got this to run and I'm able to get work done!
> Does this code have to be run on the latest Spark code? Would it run on 1.0?
> On Tuesday, July 22, 2014, Eric Garcia <no...@github.com> wrote:
>> @MLnick <https://github.com/MLnick>, I made a PR here: #1536
>> <https://github.com/apache/spark/pull/1536>
>> @rjurney <https://github.com/rjurney>, the updated code works for the
>> .avro file you posted though it is still not fully implemented for *all*
>> data types. Note that any null values in your data will show up as an empty
>> string "". For some reason I could not get Java null to convert to Python
>> None.
>>
>> —
>> Reply to this email directly or view it on GitHub
>> <https://github.com/apache/spark/pull/455#issuecomment-49792170>.
>>
> --
> Russell Jurney twitter.com/rjurney russell.jurney@gmail.com datasyndrome.com
> ---
> Reply to this email directly or view it on GitHub:
> https://github.com/apache/spark/pull/455#issuecomment-50101315
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41136741
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44794147
Regarding docs, there's now a longer section on data import in the programming guide (http://spark.apache.org/docs/latest/programming-guide.html#external-datasets), so you can add it to the Python part of that. Eventually we should probably split out data loading into a separate doc, with tutorials for HBase, Cassandra, etc.
Regarding PythonConverter, it's fine to make it experimental for now. Ideally we would look at it and keep it forever though. It's a simple enough interface (given an Object, return another Object) that we can probably keep it indefinitely, even if we later come up with a nicer / more optimized one.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45107234
Merged build finished. All automated tests passed.
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46731898
That sounds awesome, but can you put this into context a little bit, in terms of where I would put that code and how I would run it?
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45096005
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49792170
@MLnick, I made a PR here: https://github.com/apache/spark/pull/1536
@rjurney, the updated code works for the .avro file you posted though it is still not fully implemented for _all_ data types. Note that any null values in your data will show up as an empty string "". For some reason I could not get Java null to convert to Python None.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415074
--- Diff: core/src/main/scala/org/apache/spark/api/python/PythonHadoopUtil.scala ---
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.{Logging, SparkContext}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io._
+import scala.util.{Failure, Success, Try}
+
+
+trait Converter {
+ def convert(obj: Any): Any
+}
+
+object DefaultConverter extends Converter {
+
+ /**
+ * Converts a [[org.apache.hadoop.io.Writable]] to the underlying primitive, String or
+ * object representation
+ */
+ private def convertWritable(writable: Writable): Any = {
+ import collection.JavaConversions._
+ writable match {
+ case iw: IntWritable => SparkContext.intWritableConverter().convert(iw)
+ case dw: DoubleWritable => SparkContext.doubleWritableConverter().convert(dw)
+ case lw: LongWritable => SparkContext.longWritableConverter().convert(lw)
+ case fw: FloatWritable => SparkContext.floatWritableConverter().convert(fw)
+ case t: Text => SparkContext.stringWritableConverter().convert(t)
+ case bw: BooleanWritable => SparkContext.booleanWritableConverter().convert(bw)
+ case byw: BytesWritable => SparkContext.bytesWritableConverter().convert(byw)
+ case n: NullWritable => null
+ case aw: ArrayWritable => aw.get().map(convertWritable(_))
+ case mw: MapWritable => mapAsJavaMap(mw.map{ case (k, v) =>
+ (convertWritable(k), convertWritable(v))
+ }.toMap)
+ case other => other
+ }
+ }
+
+ def convert(obj: Any): Any = {
+ obj match {
+ case writable: Writable =>
+ convertWritable(writable)
+ case _ =>
+ obj
+ }
+ }
+}
+
+class ConverterRegistry extends Logging {
--- End diff --
This should probably be `private[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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13415006
--- Diff: examples/src/main/scala/org/apache/spark/examples/HBaseTest.scala ---
@@ -17,12 +17,21 @@
package org.apache.spark.examples
-import org.apache.hadoop.hbase.client.HBaseAdmin
+import org.apache.hadoop.hbase.client.{Result, HBaseAdmin}
import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor}
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.spark._
import org.apache.spark.rdd.NewHadoopRDD
+import org.apache.spark.api.python.Converter
+import org.apache.hadoop.hbase.util.Bytes
+
+class HBaseConverter extends Converter {
+ override def convert(obj: Any) = {
+ val result = obj.asInstanceOf[Result]
+ Bytes.toStringBinary(result.value())
+ }
+}
--- End diff --
Ditto on moving this to a separate 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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45094089
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148883
--- Diff: core/src/main/scala/org/apache/spark/api/python/WriteInputFormatTestDataGenerator.scala ---
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import org.apache.spark.SparkContext
+import org.apache.hadoop.io._
+import scala.Array
+import java.io.{DataOutput, DataInput}
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
+import org.apache.spark.api.java.JavaSparkContext
+
+/**
+ * A class to test MsgPack serialization on the Scala side, that will be deserialized
+ * in Python
+ * @param str
+ * @param int
+ * @param double
+ */
+case class TestWritable(var str: String, var int: Int, var double: Double) extends Writable {
+ def this() = this("", 0, 0.0)
+
+ def getStr = str
+ def setStr(str: String) { this.str = str }
+ def getInt = int
+ def setInt(int: Int) { this.int = int }
+ def getDouble = double
+ def setDouble(double: Double) { this.double = double }
+
+ def write(out: DataOutput) = {
+ out.writeUTF(str)
+ out.writeInt(int)
+ out.writeDouble(double)
+ }
+
+ def readFields(in: DataInput) = {
+ str = in.readUTF()
+ int = in.readInt()
+ double = in.readDouble()
+ }
+}
+
+/**
+ * This object contains method to generate SequenceFile test data and write it to a
+ * given directory (probably a temp directory)
+ */
+object WriteInputFormatTestDataGenerator extends App {
+ import SparkContext._
+
+ def generateData(path: String, jsc: JavaSparkContext) {
+ val sc = jsc.sc
+
+ val basePath = s"$path/sftestdata/"
+ val textPath = s"$basePath/sftext/"
+ val intPath = s"$basePath/sfint/"
+ val doublePath = s"$basePath/sfdouble/"
+ val arrPath = s"$basePath/sfarray/"
+ val mapPath = s"$basePath/sfmap/"
+ val classPath = s"$basePath/sfclass/"
+ val bytesPath = s"$basePath/sfbytes/"
+ val boolPath = s"$basePath/sfbool/"
+ val nullPath = s"$basePath/sfnull/"
+
+ /*
+ * Create test data for IntWritable, DoubleWritable, Text, BytesWritable,
+ * BooleanWritable and NullWritable
+ */
+ val intKeys = Seq((1, "aa"), (2, "bb"), (2, "aa"), (3, "cc"), (2, "bb"), (1, "aa"))
+ sc.parallelize(intKeys).saveAsSequenceFile(intPath)
+ sc.parallelize(intKeys.map{ case (k, v) => (k.toDouble, v) }).saveAsSequenceFile(doublePath)
+ sc.parallelize(intKeys.map{ case (k, v) => (k.toString, v) }).saveAsSequenceFile(textPath)
+ sc.parallelize(intKeys.map{ case (k, v) => (k, v.getBytes) }).saveAsSequenceFile(bytesPath)
+ val bools = Seq((1, true), (2, true), (2, false), (3, true), (2, false), (1, false))
+ sc.parallelize(bools).saveAsSequenceFile(boolPath)
+ sc.parallelize(intKeys).map{ case (k, v) =>
+ (new IntWritable(k), NullWritable.get())
+ }.saveAsSequenceFile(nullPath)
+
+ // Create test data for ArrayWritable
+ val data = Seq(
+ (1, Array(1.0, 2.0, 3.0)),
+ (2, Array(3.0, 4.0, 5.0)),
+ (3, Array(4.0, 5.0, 6.0))
+ )
+ sc.parallelize(data, numSlices = 2)
+ .map{ case (k, v) =>
+ (new IntWritable(k), new ArrayWritable(classOf[DoubleWritable], v.map(new DoubleWritable(_))))
+ }
+ .saveAsNewAPIHadoopFile[SequenceFileOutputFormat[IntWritable, ArrayWritable]](arrPath)
--- End diff --
Formatting here is weird, perhaps due to tabs. It would look better as
```
sc.parallelize(data, numSlices = 2)
.map { case (k, v) =>
(new IntWritable(k), new ArrayWritable(classOf[DoubleWritable], v.map(new DoubleWritable(_))))
}.saveAsNewAPIHadoopFile[SequenceFileOutputFormat[IntWritable, ArrayWritable]](arrPath)
```
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44450061
Hey Nick, I finally had time to do a first pass through this. This looks very good -- it's great how broad a set of classes it supports out of the box. I made some notes on the PR, but also a few high-level comments:
* It would be good to move the sequence file tests to `tests.py` instead of doctests since these are fairly involved.
* `hadoopRDD`, `hadoopFile`, etc also need tests. You can pass SequenceFileInputFormat or TextInputFormat to those and just check that they work.
* It would be good to add a few tests where the key or value class passed is wrong, just to check that we have a sensible error message and don't completely crash.
* What are your thoughts on letting people specify their own wrappers? Maybe we can make a Java / Scala interface called `PythonConverter` or something that users can implement, and then they'd pass a class name for a custom PythonConverter. We should just make sure this is a stable API we can support in the future.
* I like automatically inferring the data type for SequenceFiles, as it's very Python-like, but maybe we shouldn't do it for other types of hadoop datasets at first just in case those InputFormats need to have the key and value class set in the Configuration. I'm not 100% sure how this works though (whether some InputFormats look at those). SequenceFile has that info in the file itself and ignores what you pass in.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41909611
@mateiz any chance to look into merging this, or comments / questions?
Sent from my iPhone
> On 23 Apr 2014, at 22:56, UCB AMPLab <no...@github.com> wrote:
>
> Merged build finished. All automated tests passed.
>
> —
> Reply to this email directly or view it on GitHub.
---
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46745394
@rjurney this works for me (building Spark from current master): https://gist.github.com/MLnick/5864741781b9340cb211
if you run ```mvn package``` and then add that to ```SPARK_CLASSPATH``` and use it in IPython console.
However it seems to come through as only strings (not a dict). I verified that if I take only the string field and explicitly convert to string (ie ```Map[String, String]```) then it works. I suspect then that Avro doesn't have the type information at all, so Pyrolite cannot pickle it. I guess you might have to do something more in depth in the ```AvroConverter``` to read the type info from the Avro schema and do a cast...
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40973978
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14301/
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41208390
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-1416: PySpark support for SequenceFile a...
Posted by pwendell <gi...@git.apache.org>.
Github user pwendell commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45415893
So after looking at this more. I think it's great to add to support for sequence file format in Python, it will be super useful.
This patch takes things a step further and also introduces an API for users to plug in different types of Hadoop input formats. I'm a bit concerned about exposing that to users (which this patch does via the `Converter` stuff) because I'm sure people will start building on it and it's not an interface I'd be happy supporting going forward. For one thing, it only applies to Python. Another thing is that there are other outstanding community proposals for how to deal with things like HBase support (see #194). In the near future (possibly 1.1) we're planning to standardize the way this works via SchemaRDD, and that will be automatically supported in Python. Basically, we'll have ways to read and write to a SchemaRDD from several storage systems.
So I just don't think it's a good idea to introduce a different pluggable mechanism here and encourage users to integrate at this level, and I wonder if we should just merge a version of this patch that only exposes sequenceFile and not the more general converter mechanism.
@mateiz @marmbrus curious to hear your thoughts as well
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-44450146
BTW if we add a converter interface it would be nice to rename those parameters to `key_converter` and `value_converter`.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13147863
--- Diff: python/pyspark/context.py ---
@@ -311,6 +311,98 @@ def wholeTextFiles(self, path, minPartitions=None):
return RDD(self._jsc.wholeTextFiles(path, minPartitions), self,
PairDeserializer(UTF8Deserializer(), UTF8Deserializer()))
+ def sequenceFile(self, name, key_class="org.apache.hadoop.io.Text", value_class="org.apache.hadoop.io.Text",
+ key_wrapper="", value_wrapper="", minSplits=None):
+ """
+ Read a Hadoop SequenceFile with arbitrary key and value Writable class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is as follows:
+ 1. A Java RDD is created from the SequenceFile or other InputFormat, and the key and value Writable classes
+ 2. Serialization is attempted via Pyrolite pickling
+ 3. If this fails, the fallback is to call 'toString' on each key and value
+ 4. C{PickleSerializer} is used to deserialize pickled objects on the Python side
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfint/").collect())
+ [(1, u'aa'), (1, u'aa'), (2, u'aa'), (2, u'bb'), (2, u'bb'), (3, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfdouble/").collect())
+ [(1.0, u'aa'), (1.0, u'aa'), (2.0, u'aa'), (2.0, u'bb'), (2.0, u'bb'), (3.0, u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sftext/").collect())
+ [(u'1', u'aa'), (u'1', u'aa'), (u'2', u'aa'), (u'2', u'bb'), (u'2', u'bb'), (u'3', u'cc')]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfbool/").collect())
+ [(1, False), (1, True), (2, False), (2, False), (2, True), (3, True)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfnull/").collect())
+ [(1, None), (1, None), (2, None), (2, None), (2, None), (3, None)]
+ >>> sorted(sc.sequenceFile(tempdir + "/sftestdata/sfmap/").collect())
+ [(1, {2.0: u'aa'}), (1, {3.0: u'bb'}), (2, {1.0: u'aa'}), (2, {1.0: u'cc'}), (2, {3.0: u'bb'}), (3, {2.0: u'dd'})]
+ >>> r = sorted(sc.sequenceFile(tempdir + "/sftestdata/sfclass").collect())[0]
+ >>> r == (u'1', {u'__class__': u'org.apache.spark.api.python.TestWritable', u'double': 54.0, u'int': 123, u'str': u'test1'})
+ True
+ """
+ minSplits = minSplits or min(self.defaultParallelism, 2)
+ jrdd = self._jvm.PythonRDD.sequenceFile(self._jsc, name, key_class, value_class, key_wrapper, value_wrapper,
+ minSplits)
+ return RDD(jrdd, self, PickleSerializer())
+
+ def newAPIHadoopFile(self, name, inputformat_class, key_class="org.apache.hadoop.io.Text",
+ value_class="org.apache.hadoop.io.Text", key_wrapper="toString",
+ value_wrapper="toString", conf={}):
+ """
+ Read a 'new API' Hadoop InputFormat with arbitrary key and value class from HDFS,
+ a local file system (available on all nodes), or any Hadoop-supported file system URI.
+ The mechanism is the same as for sc.sequenceFile.
+
+ A Hadoop configuration can be passed in as a Python dict. This will be converted into a Configuration in Java
+ """
+ jconf = self._jvm.java.util.HashMap()
+ for k, v in conf.iteritems():
+ jconf[k] = v
+ jrdd = self._jvm.PythonRDD.newAPIHadoopFile(self._jsc, name, inputformat_class, key_class, value_class,
+ key_wrapper, value_wrapper, jconf)
+ return RDD(jrdd, self, PickleSerializer())
+
+ def newAPIHadoopRDD(self, inputformat_class, key_class="org.apache.hadoop.io.Text",
+ value_class="org.apache.hadoop.io.Text", key_wrapper="", value_wrapper="", conf={}):
--- End diff --
For all of these methods I'd suggest not having a default key_class and value_class. Did you find it inconvenient to always pass them? I'm worried that a user will miss them and then get hard to explain ClassCastExceptions. It seems like a small ask to make these required parameters by default.
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13148379
--- Diff: core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala ---
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import scala.util.Try
+import org.apache.spark.rdd.RDD
+import org.apache.spark.Logging
+import scala.util.Success
+import scala.util.Failure
+import net.razorvine.pickle.Pickler
+
+/**
+ * Utilities for serialization / deserialization between Python and Java, using MsgPack.
+ * Also contains utilities for converting [[org.apache.hadoop.io.Writable]] ->
+ * Scala objects and primitives
+ */
+private[python] object SerDeUtil extends Logging {
+
+ /**
+ * Convert an RDD of key-value pairs to an RDD of serialized Python objects, that is usable
+ * by PySpark. By default, if serialization fails, toString is called and the string
+ * representation is serialized
+ */
+ def rddToPython[K, V](rdd: RDD[(K, V)]): RDD[Array[Byte]] = {
+ rdd.mapPartitions{ iter =>
+ val pickle = new Pickler
+ var keyFailed = false
+ var valueFailed = false
+ var firstRecord = true
+ iter.map{ case (k, v) =>
+ if (firstRecord) {
+ Try {
+ pickle.dumps(Array(k, v))
+ } match {
+ case Success(b) =>
+ case Failure(err) =>
+ val kt = Try {
+ pickle.dumps(k)
+ }
+ val vt = Try {
+ pickle.dumps(v)
+ }
+ (kt, vt) match {
+ case (Failure(kf), Failure(vf)) =>
+ log.warn(s"""Failed to pickle Java object as key: ${k.getClass.getSimpleName};
+ Error: ${kf.getMessage}""")
+ log.warn(s"""Failed to pickle Java object as value: ${v.getClass.getSimpleName};
+ Error: ${vf.getMessage}""")
+ keyFailed = true
+ valueFailed = true
+ case (Failure(kf), _) =>
+ log.warn(s"""Failed to pickle Java object as key: ${k.getClass.getSimpleName};
+ Error: ${kf.getMessage}""")
+ keyFailed = true
+ case (_, Failure(vf)) =>
+ log.warn(s"""Failed to pickle Java object as value: ${v.getClass.getSimpleName};
+ Error: ${vf.getMessage}""")
+ valueFailed = true
+ }
+ }
+ firstRecord = false
+ }
+ (keyFailed, valueFailed) match {
+ case (true, true) => pickle.dumps(Array(k.toString, v.toString))
+ case (true, false) => pickle.dumps(Array(k.toString, v))
+ case (false, true) => pickle.dumps(Array(k, v.toString))
+ case (false, false) => pickle.dumps(Array(k, v))
+ }
--- End diff --
I'd replace this with `if` statements here just because a pattern-match on every record is fairly expensive. But it's a minor thing.
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45320861
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r13426219
--- Diff: examples/src/main/scala/org/apache/spark/examples/CassandraCQLTest.scala ---
@@ -32,6 +32,23 @@ import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.SparkContext._
+import org.apache.spark.api.python.Converter
+
+class CassandraCQLKeyConverter extends Converter {
+ import collection.JavaConversions._
+ override def convert(obj: Any) = {
+ val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]]
+ mapAsJavaMap(result.mapValues(bb => ByteBufferUtil.toInt(bb)))
+ }
+}
+
+class CassandraCQLValueConverter extends Converter {
+ import collection.JavaConversions._
+ override def convert(obj: Any) = {
+ val result = obj.asInstanceOf[java.util.Map[String, ByteBuffer]]
+ mapAsJavaMap(result.mapValues(bb => ByteBufferUtil.string(bb)))
+ }
+}
--- End diff --
will do
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878606
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14258/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878576
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41213713
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/14392/
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41204439
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-1416: PySpark support for SequenceFile a...
Posted by ericgarcia <gi...@git.apache.org>.
Github user ericgarcia commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49681825
@rjurney Thanks for your question on Avro and pyspark. @MLnick I used your code as a starting point to successfully read a few different Avro data types in pyspark instead of merely a String. This could be added in with HBaseConverter.scala and CassandraConverters.scala
package org.apache.spark.examples.pythonconverters
import org.apache.spark.api.python.Converter
import org.apache.avro.generic.GenericRecord
import org.apache.avro.mapred.AvroKey
import org.apache.avro.mapreduce.AvroKeyInputFormat
import collection.JavaConversions._
import org.apache.avro.Schema.Field
import org.apache.avro.Schema
import org.apache.avro.Schema.Type._
class AvroGenericConverter extends Converter[AvroKey[GenericRecord], java.util.Map[String, Any]] {
override def convert(obj: AvroKey[GenericRecord]): java.util.Map[String, Any] = {
val record = obj.datum()
def unpackRecord(record: GenericRecord): java.util.Map[String,Any] = {
mapAsJavaMap(record.getSchema.getFields.map( f => (f.name, unpack(record.get(f.name), f.schema) ) ).toMap)
}
def unpackArray(value: Any, schema: Schema): java.util.List[Any] = {
bufferAsJavaList(value.asInstanceOf[java.util.List[Any]].map( v => unpack(v, schema)))
}
def unpack(value: Any, schema: Schema): Any = schema.getType match {
case STRING => value.asInstanceOf[java.lang.String]
case ENUM => value.toString
case LONG => value.asInstanceOf[java.lang.Long]
case INT => value.asInstanceOf[java.lang.Integer]
case ARRAY => unpackArray(value, schema.getElementType)
case RECORD => unpackRecord(value.asInstanceOf[GenericRecord])
case _ => value.toString
}
unpackRecord(record)
}
}
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-50420550
@ericgarcia I am running CDH 4.4 with mapreduce 1, and when I run this on a cluster running Spark master in standalone mode:
avroRdd = sc.newAPIHadoopFile("hdfs://hivecluster2:8020/e8/dev/web_proxy_mef/2014/07/28",
"org.apache.avro.mapreduce.AvroKeyInputFormat",
"org.apache.avro.mapred.AvroKey",
"org.apache.hadoop.io.NullWritable",
keyConverter="org.apache.spark.examples.pythonconverters.AvroGenericConverter")
The error is the well known hadoop 1 vs hadoop 2 issue.
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.newAPIHadoopFile.
: org.apache.hadoop.ipc.RemoteException: Server IPC version 7 cannot communicate with client version 4
So I also tried hadoopFile, instead of newAPIHadoopFile. I run:
avroRdd = sc.hadoopFile("hdfs://hivecluster2:8020/e8/dev/web_proxy_mef/2014/07/28/20/part-m-00000.avro",
"org.apache.avro.mapreduce.AvroKeyInputFormat",
"org.apache.avro.mapred.AvroKey",
"org.apache.hadoop.io.NullWritable",
keyConverter="org.apache.spark.examples.pythonconverters.AvroGenericConverter")
And I get:
Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.hadoopFile.
: java.lang.ClassCastException: org.apache.avro.mapreduce.AvroKeyInputFormat cannot be cast to org.apache.hadoop.mapred.InputFormat
What am I to do? :(
---
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-1416: PySpark support for SequenceFile a...
Posted by rjurney <gi...@git.apache.org>.
Github user rjurney commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-49698081
It also looks like we need a custom function to handle the UNION type. I've extended what you wrote for DOUBLE/FLOAT:
def unpack(value: Any, schema: Schema): Any = schema.getType match {
case STRING => value.asInstanceOf[java.lang.String]
case ENUM => value.toString
case LONG => value.asInstanceOf[java.lang.Long]
case INT => value.asInstanceOf[java.lang.Integer]
case FLOAT => value.asInstanceOf[java.lang.Float]
case DOUBLE => value.asInstanceOf[java.lang.Double]
case ARRAY => unpackArray(value, schema.getElementType)
case RECORD => unpackRecord(value.asInstanceOf[GenericRecord])
case _ => value.toString
}
---
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on a diff in the pull request:
https://github.com/apache/spark/pull/455#discussion_r11919159
--- Diff: python/pyspark/serializers.py ---
@@ -114,11 +113,6 @@ class FramedSerializer(Serializer):
where C{length} is a 32-bit integer and data is C{length} bytes.
"""
- def __init__(self):
- # On Python 2.6, we can't write bytearrays to streams, so we need to convert them
- # to strings first. Check if the version number is that old.
- self._only_write_strings = sys.version_info[0:2] <= (2, 6)
--- End diff --
Why was this removed? It seems you did a git merge or rebase from an earlier version and lost some fixes added in-between
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45411710
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45365824
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15511/
---
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45101273
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45344727
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-1416: PySpark support for SequenceFile a...
Posted by kanzhang <gi...@git.apache.org>.
Github user kanzhang commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45827287
@MLnick thanks for the tip, Nick! I'll give it a try and ask you to review.
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878970
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-1416: PySpark support for SequenceFile a...
Posted by mateiz <gi...@git.apache.org>.
Github user mateiz commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-41189277
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-1416: PySpark support for SequenceFile a...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-45107235
All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15448/
---
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: [WIP] SPARK-1416: PySpark support for Sequence...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-40878570
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-1416: PySpark support for SequenceFile a...
Posted by MLnick <gi...@git.apache.org>.
Github user MLnick commented on the pull request:
https://github.com/apache/spark/pull/455#issuecomment-46810658
hmmm not sure - master built fine for me at the time I posted. Either try pull again or maybe checkout my branch and try that: https://github.com/MLnick/spark-1/tree/pyspark-inputformats
---
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.
---