Eager serialize values of ParallelCollection

douban · Dec 5, 2016 · 94c5ef6 · 94c5ef6
1 parent fef3de9
commit 94c5ef6
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/dpark/rdd.py b/dpark/rdd.py
@@ -1089,7 +1089,7 @@ def compute(self, split):
 class ParallelCollectionSplit:
     def __init__(self, index, values):
         self.index = index
-        self.values = values
+        self.values = cPickle.dumps(values, -1)
 
 class ParallelCollection(RDD):
     def __init__(self, ctx, data, numSlices, taskMemory=None):
@@ -1104,7 +1104,7 @@ def __init__(self, ctx, data, numSlices, taskMemory=None):
         self.repr_name = '<ParallelCollection %d>' % self.size
 
     def compute(self, split):
-        return split.values
+        return cPickle.loads(split.values)
 
     @classmethod
     def slice(cls, data, numSlices):