My intention is to add pyspark support for certain mllib spark methods. I have been unable to resolve pickling errors of the form
Pyspark py4j PickleException: “expected zero arguments for construction of ClassDict” <http://stackoverflow.com/questions/29910708/pyspark-py4j-pickleexception-expected-zero-arguments-for-construction-of-class> These are occurring during python to java conversion of python named tuples. The details are rather hard to provide here so I have created an SOF question http://stackoverflow.com/questions/29910708/pyspark-py4j-pickleexception-expected-zero-arguments-for-construction-of-class In any case I have included the text here. The SOF is easier to read though ;) -------------- This question is directed towards persons familiar with py4j - and can help to resolve a pickling error. I am trying to add a method to the pyspark PythonMLLibAPI that accepts an RDD of a namedtuple, does some work, and returns a result in the form of an RDD. This method is modeled after the PYthonMLLibAPI.trainALSModel() method, whose analogous *existing* relevant portions are: def trainALSModel( ratingsJRDD: JavaRDD[Rating], .. ) The *existing* python Rating class used to model the new code is: class Rating(namedtuple("Rating", ["user", "product", "rating"])): def __reduce__(self): return Rating, (int(self.user), int(self.product), float(self.rating)) Here is the attempt So here are the relevant classes: *New* python class pyspark.mllib.clustering.MatrixEntry: from collections import namedtupleclass MatrixEntry(namedtuple("MatrixEntry", ["x","y","weight"])): def __reduce__(self): return MatrixEntry, (long(self.x), long(self.y), float(self.weight)) *New* method *foobarRDD* In PythonMLLibAPI: def foobarRdd( data: JavaRDD[MatrixEntry]): RDD[FooBarResult] = { val rdd = data.rdd.map { d => FooBarResult(d.i, d.j, d.value, d.i * 100 + d.j * 10 + d.value)} rdd } Now let us try it out: from pyspark.mllib.clustering import MatrixEntry def convert_to_MatrixEntry(tuple): return MatrixEntry(*tuple) from pyspark.mllib.clustering import * pic = PowerIterationClusteringModel(2) tups = [(1,2,3),(4,5,6),(12,13,14),(15,7,8),(16,17,16.5)] trdd = sc.parallelize(map(convert_to_MatrixEntry,tups)) # print out the RDD on python side just for validationprint "%s" %(repr(trdd.collect())) from pyspark.mllib.common import callMLlibFunc pic = callMLlibFunc("foobar", trdd) Relevant portions of results: [(1,2)=3.0, (4,5)=6.0, (12,13)=14.0, (15,7)=8.0, (16,17)=16.5] which shows the input rdd is 'whole'. However the pickling was unhappy: 5/04/27 21:15:44 ERROR Executor: Exception in task 6.0 in stage 1.0 (TID 14) net.razorvine.pickle.PickleException: expected zero arguments for construction of ClassDict(for pyspark.mllib.clustering.MatrixEntry) at net.razorvine.pickle.objects.ClassDictConstructor.construct(ClassDictConstructor.java:23) at net.razorvine.pickle.Unpickler.load_reduce(Unpickler.java:617) at net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:170) at net.razorvine.pickle.Unpickler.load(Unpickler.java:84) at net.razorvine.pickle.Unpickler.loads(Unpickler.java:97) at org.apache.spark.mllib.api.python.SerDe$$anonfun$pythonToJava$1$$anonfun$apply$1.apply(PythonMLLibAPI.scala:1167) at org.apache.spark.mllib.api.python.SerDe$$anonfun$pythonToJava$1$$anonfun$apply$1.apply(PythonMLLibAPI.scala:1166) at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327) 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$17.apply(RDD.scala:819) at org.apache.spark.rdd.RDD$$anonfun$17.apply(RDD.scala:819) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1523) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1523) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) at org.apache.spark.scheduler.Task.run(Task.scala:64) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:212) 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) Here is the python invocation stack trace: Py4JJavaError Traceback (most recent call last) <ipython-input-2-3589950a5c09> in <module>() 12 13 from pyspark.mllib.common import callMLlibFunc ---> 14 pic = callMLlibFunc("foobar", trdd) /shared/picpy/python/pyspark/mllib/common.pyc in callMLlibFunc(name, *args) 119 sc = SparkContext._active_spark_context 120 api = getattr(sc._jvm.PythonMLLibAPI(), name) --> 121 return callJavaFunc(sc, api, *args) 122 123 /shared/picpy/python/pyspark/mllib/common.pyc in callJavaFunc(sc, func, *args) 112 """ Call Java Function """ 113 args = [_py2java(sc, a) for a in args] --> 114 return _java2py(sc, func(*args)) 115 116 /Library/Python/2.7/site-packages/py4j-0.8.2.1-py2.7.egg/py4j/java_gateway.pyc in __call__(self, *args) 536 answer = self.gateway_client.send_command(command) 537 return_value = get_return_value(answer, self.gateway_client, --> 538 self.target_id, self.name) 539 540 for temp_arg in temp_args: /Library/Python/2.7/site-packages/py4j-0.8.2.1-py2.7.egg/py4j/protocol.pyc 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 o31.foobar.
