def _jrdd(self):
     if self._jrdd_val:
         return self._jrdd_val
     func = self.func
     if not self._bypass_serializer and self.ctx.batchSize != 1:
         oldfunc = self.func
         batchSize = self.ctx.batchSize
         def batched_func(split, iterator):
             return batched(oldfunc(split, iterator), batchSize)
         func = batched_func
     cmds = [func, self._bypass_serializer]
     pipe_command = ' '.join(b64enc(cloudpickle.dumps(f)) for f in cmds)
     broadcast_vars = ListConverter().convert(
         [x._jbroadcast for x in self.ctx._pickled_broadcast_vars],
         self.ctx._gateway._gateway_client)
     self.ctx._pickled_broadcast_vars.clear()
     class_manifest = self._prev_jrdd.classManifest()
     env = MapConverter().convert(self.ctx.environment,
                                  self.ctx._gateway._gateway_client)
     includes = ListConverter().convert(self.ctx._python_includes,
                                  self.ctx._gateway._gateway_client)
     python_rdd = self.ctx._jvm.PythonRDD(self._prev_jrdd.rdd(),
         pipe_command, env, includes, self.preservesPartitioning, self.ctx.pythonExec,
         broadcast_vars, self.ctx._javaAccumulator, class_manifest)
     self._jrdd_val = python_rdd.asJavaRDD()
     return self._jrdd_val

 def dumps(self, obj):
     try:
         return cloudpickle.dumps(obj, 2)
     except pickle.PickleError:
         raise
     except Exception as e:
         emsg = _exception_message(e)
         if "'i' format requires" in emsg:
             msg = "Object too large to serialize: %s" % emsg
         else:
             msg = "Could not serialize object: %s: %s" % (e.__class__.__name__, emsg)
         cloudpickle.print_exec(sys.stderr)
         raise pickle.PicklingError(msg)

 def dumps(self, obj):
     return cloudpickle.dumps(obj, 2)

    def dumps(self, obj): return cloudpickle.dumps(obj, 2)


class MarshalSerializer(FramedSerializer):