Pyspark py4j PickleException: "dự kiến ​​không đối số để xây dựng ClassDict"

Question

Câu hỏi này hướng tới những người quen thuộc với py4j - và có thể giúp giải quyết một lỗi tẩy. Tôi đang cố gắng để thêm một phương pháp để các pyspark PythonMLLibAPI chấp nhận một RDD của một namedtuple, hiện một số công việc, và trả về một kết quả trong các hình thức của một RDD.

Phương pháp này được mô hình hóa sau khi phương pháp PYthonMLLibAPI.trainALSModel(), mà hiện phần có liên quan tương tự là:

def trainALSModel(
    ratingsJRDD: JavaRDD[Rating], 
    ..) 

Các hiện python lớp Đánh giá sử dụng để mô hình mã mới là:

class Rating(namedtuple("Rating", ["user", "product", "rating"])): 
    def __reduce__(self): 
     return Rating, (int(self.user), int(self.product), float(self.rating)) 

Đây là nỗ lực Vì vậy, đây là các lớp có liên quan:

.210

New python lớp pyspark.mllib.clustering.MatrixEntry:

from collections import namedtuple 
class MatrixEntry(namedtuple("MatrixEntry", ["x","y","weight"])): 
    def __reduce__(self): 
     return MatrixEntry, (long(self.x), long(self.y), float(self.weight)) 

New phương pháp foobarRDD Trong 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 
    } 

Bây giờ chúng ta hãy thử nó ra:

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 validation 
print "%s" %(repr(trdd.collect())) 

from pyspark.mllib.common import callMLlibFunc 
pic = callMLlibFunc("foobar", trdd) 

Các phần có liên quan của lại sults:

[(1,2)=3.0, (4,5)=6.0, (12,13)=14.0, (15,7)=8.0, (16,17)=16.5] 

hiển thị phần nhập là "toàn bộ". Tuy nhiên, tẩy không hài lòng:

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) 

Dưới đây là hình ảnh của python gọi stack trace:

Answer 1

tôi đã có lỗi tương tự như tôi đã sử dụng MLlib, và nó bật ra rằng tôi đã trả lại một kiểu dữ liệu sai ở một trong các chức năng của tôi. Nó bây giờ hoạt động sau khi một diễn viên đơn giản trên giá trị trả lại. Đây có thể không phải là câu trả lời bạn đang tìm kiếm nhưng ít nhất cũng là gợi ý cho hướng đi theo.

Answer 2

Tôi đã nhận được lỗi này khi sử dụng Phiên bản Spark> = 2.0.

Spark đang chuyển đổi tính tương đồng MLlib thành không gian tên ML mới hơn. Kết quả là có hai loại SparseVector: ml.linalg.SparseVector và mllib.linalg.SparseVector

Một số chức năng MLlib vẫn mong đợi mllib loại cũ

from pyspark.ml.linalg import Vectors 
# convert ML vector to older MLlib vector 
old_vec = Vectors.fromML(new_vec) 

HTH

Answer 3

Đã cùng một vấn đề, nhiều lần. numpy các loại không có chuyển đổi tiềm ẩn cho pyspark.sql.types.

Thực hiện chuyển đổi rõ ràng đơn giản sang hệ thống kiểu gốc. Trong trường hợp của tôi đó là:

float(vector_a.dot(vector_b)