SPARK-1416: PySpark support for SequenceFile and Hadoop InputFormats

[MLnick]

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 SequenceFiles, as well as arbitrary Hadoop InputFormats, 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. Pyrolite is used to serialize Java objects. If this fails, the fallback is toString
4. PickleSerializer on the Python side deserializes.

This works "out the box" for simple Writables:

• Text
• IntWritable, DoubleWritable, FloatWritable
• NullWritable
• BooleanWritable
• BytesWritable
• MapWritable

It also works for simple, "struct-like" classes. Due to the way Pyrolite works, this requires that the classes satisfy the JavaBeans convenstions (i.e. with fields and a no-arg constructor and getters/setters). (Perhaps in future some sugar for case classes and reflection could be added).

I've tested it out with ESInputFormat as an example and it works very nicely:

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 SPARK-1374: PySpark API for SparkSQL #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 JavaBean or java.util.Map that can be easily serialized)
4. Support saveAsSequenceFile and saveAsHadoopFile etc. This would require SerDe in the reverse direction, that can be handled by Pyrolite. Will work on this as a separate PR

[rjurney]

@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

[rjurney]

@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

[srowen]

You're not literally writing 4.X.X in the version are you?

[rjurney]

@srowen Actually yes, I'm that stupid :) Figured it out on me own though, have it building across the cluster now.

[rjurney]

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.(NettyTransport.scala:282)
at akka.remote.transport.netty.NettyTransport.(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)

[rjurney]

Hmmm, I didn't clean before rebuilding with CDH 4.4. Trying that now.

[rjurney]

@srowen @JoshRosen Cleaned, made no difference. See https://issues.apache.org/jira/browse/SPARK-1138 where others had this issue.

[srowen]

@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.

[ericgarcia]

@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.

[rjurney]

@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

Which looks like it isn't applying?

[srowen]

@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.

[rjurney]

@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.

[ericgarcia]

@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")

[rjurney]

@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?

[MLnick]

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 notifications@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:
#455 (comment)

[freedafeng]

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.

[JoshRosen]

@freedafeng This PR was actually merged and will be available in Spark 1.1 (which should be released very soon). f971d6c