Return an JavaRDD of Object by unpickling

    It will convert each Python object into Java object by Pyrolite, whenever the
    RDD is serialized in batch or not.
965126
def _to_java_object_rdd(rdd):
    
    rdd = rdd._reserialize(AutoBatchedSerializer(PickleSerializer()))
    return rdd.ctx._jvm.org.apache.spark.ml.python.MLSerDe.pythonToJava(rdd._jrdd, True)
1068968
def serialize(obj, **options):
    '''
    
    '''

    #round-trip this through JSON to avoid OrderedDict types
    # there's probably a more performant way to do this...
    # TODO remove json round-trip when all dataset will use
    # serializers
    return pprint.pformat(
        salt.utils.json.loads(
            salt.utils.json.dumps(obj, _json_module=_json),
            _json_module=_json
        ),
        **options
    )
378339
function() {

        this.logger = Logger.getLogger("FreqItemset_js");
        var jvmObject = arguments[0];
        if (arguments.length > 1) {
            var items = Utils.unwrapObject(arguments[0]);
            if (Array.isArray(items)) {
                var list = new java.util.ArrayList();
                items.forEach(function(item){
                    list.add(Utils.unwrapObject(item));
                });
                items = list.toArray();
            }
            jvmObject = new org.apache.spark.mllib.fpm.FPGrowth.FreqItemset(items, arguments[1]);
        }
        JavaWrapper.call(this, jvmObject);

    }
216187
def foreachRDD(self, func):
        
        if func.__code__.co_argcount == 1:
            old_func = func
            func = lambda t, rdd: old_func(rdd)
        jfunc = TransformFunction(self._sc, func, self._jrdd_deserializer)
        api = self._ssc._jvm.PythonDStream
        api.callForeachRDD(self._jdstream, jfunc)
507050
function (jvmObject) {

        this.logger = Logger.getLogger("VectorUDT_js");
        if (!jvmObject) {
            jvmObject = new org.apache.spark.ml.linalg.VectorUDT();
        }
        JavaWrapper.call(this, jvmObject);

    }
607131
def persistent_load(self, pid):
        
        try:
            type_tag, key_id = pid
        except Exception as exc:
            # Sometimes we get a string such as ('Element', u'C') instead
            # of a real tuple. Use ast to evalute the expression (much safer
            # than eval).
            import ast
            type_tag, key_id = ast.literal_eval(pid)

        if type_tag == "Element":
            return Element(key_id)
        else:
            # Always raises an error if you cannot return the correct object.
            # Otherwise, the unpickler will think None is the object referenced
            # by the persistent ID.
            raise pickle.UnpicklingError(
                "unsupported persistent object with pid %s" % pid)
383672
def _to_java(self):
        
        sc = SparkContext._active_spark_context
        java_models = [model._to_java() for model in self.models]
        java_models_array = JavaWrapper._new_java_array(
            java_models, sc._gateway.jvm.org.apache.spark.ml.classification.ClassificationModel)
        metadata = JavaParams._new_java_obj("org.apache.spark.sql.types.Metadata")
        _java_obj = JavaParams._new_java_obj("org.apache.spark.ml.classification.OneVsRestModel",
                                             self.uid, metadata.empty(), java_models_array)
        _java_obj.set("classifier", self.getClassifier()._to_java())
        _java_obj.set("featuresCol", self.getFeaturesCol())
        _java_obj.set("labelCol", self.getLabelCol())
        _java_obj.set("predictionCol", self.getPredictionCol())
        return _java_obj
524105
def unmarshal_event(self, data: str, response_type):
        
        js = json.loads(data)

        # Make a copy of the original object and save it under the
        # `raw_object` key because we will replace the data under `object` with
        # a Python native type shortly.
        js['raw_object'] = js['object']

        # Something went wrong. A typical example would be that the user
        # supplied a resource version that was too old. In that case K8s would
        # not send a conventional ADDED/DELETED/... event but an error. Turn
        # this error into a Python exception to save the user the hassle.
        if js['type'].lower() == 'error':
            return js

        # If possible, compile the JSON response into a Python native response
        # type, eg `V1Namespace` or `V1Pod`,`ExtensionsV1beta1Deployment`, ...
        if response_type is not None:
            js['object'] = self._api_client.deserialize(
                response=SimpleNamespace(data=json.dumps(js['raw_object'])),
                response_type=response_type
            )

            # decode and save resource_version to continue watching
            if hasattr(js['object'], 'metadata'):
                self.resource_version = js['object'].metadata.resource_version

            # For custom objects that we don't have model defined, json
            # deserialization results in dictionary
            elif (isinstance(js['object'], dict) and
                  'metadata' in js['object'] and
                  'resourceVersion' in js['object']['metadata']):

                self.resource_version = js['object']['metadata']['resourceVersion']

        return js
1135437
def _loads(self, string):
        
        if not self.serialized:
            return self._decode(string)
        if string is not None:
            try:
                return self.serializer.loads(string)
            except TypeError:
                #: catches bytes errors with the builtin json library
                return self.serializer.loads(self._decode(string))
            except pickle.UnpicklingError as e:
                #: incr and decr methods create issues when pickle serialized
                #  It's a terrible idea for a serialized instance
                #  to be performing incr and decr methods, but I think
                #  it makes sense to catch the error regardless
                decoded = self._decode(string)
                if decoded.isdigit():
                    return decoded
                raise pickle.UnpicklingError(e)
811414
def to_python(self):
        

        result = None
        if self.type == PythonTypes.null:
            result = None
        elif self.type == PythonTypes.bool:
            result = self.value == 1
        elif self.type == PythonTypes.integer:
            if self.value is None:
                result = 0
            else:
                result = ctypes.c_int32(self.value).value
        elif self.type == PythonTypes.double:
            result = self._double_value()
        elif self.type == PythonTypes.str_utf8:
            buf = ctypes.c_char_p(self.value)
            ptr = ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))
            result = ptr[0:self.len].decode("utf8")
        elif self.type == PythonTypes.array:
            if self.len == 0:
                return []
            ary = []
            ary_addr = ctypes.c_void_p.from_address(self.value)
            ptr_to_ary = ctypes.pointer(ary_addr)
            for i in range(self.len):
                pval = PythonValue.from_address(ptr_to_ary[i])
                ary.append(pval.to_python())
            result = ary
        elif self.type == PythonTypes.hash:
            if self.len == 0:
                return {}
            res = {}
            hash_ary_addr = ctypes.c_void_p.from_address(self.value)
            ptr_to_hash = ctypes.pointer(hash_ary_addr)
            for i in range(self.len):
                pkey = PythonValue.from_address(ptr_to_hash[i*2])
                pval = PythonValue.from_address(ptr_to_hash[i*2+1])
                res[pkey.to_python()] = pval.to_python()
            result = res
        elif self.type == PythonTypes.function:
            result = JSFunction()
        elif self.type == PythonTypes.parse_exception:
            msg = ctypes.c_char_p(self.value).value
            raise JSParseException(msg)
        elif self.type == PythonTypes.execute_exception:
            msg = ctypes.c_char_p(self.value).value
            raise JSEvalException(msg.decode('utf-8', errors='replace'))
        elif self.type == PythonTypes.oom_exception:
            msg = ctypes.c_char_p(self.value).value
            raise JSOOMException(msg)
        elif self.type == PythonTypes.timeout_exception:
            msg = ctypes.c_char_p(self.value).value
            raise JSTimeoutException(msg)
        elif self.type == PythonTypes.date:
            timestamp = self._double_value()
            # JS timestamp are milliseconds, in python we are in seconds
            result = datetime.datetime.utcfromtimestamp(timestamp / 1000.)
        else:
            raise WrongReturnTypeException("unknown type %d" % self.type)
        return result
1372809
def of_pyobj(self, pyobj):
        
        m = self.hash_algo()
        m.update(pickle.dumps(pyobj, protocol=self.pk_protocol))
        return self.digest(m)
139459
def to_python(self, value):
        """"""
        cache = None
        if value is None:
            return None

        if isinstance(value, DataDescriptor):
            return value
        elif isinstance(value, dict):
            # Allow pre-hydrated data objects.
            cache = value
            try:
                value = cache['__id']
            except KeyError:
                raise ValidationError("dictionary must contain an '__id' element")
        elif not isinstance(value, int):
            raise ValidationError("field must be a DataDescriptor, data object primary key or dict")

        return DataDescriptor(value, self, cache)
1118588
def _iter_convert_to_object(self, iterable):
        """"""
        for binsha, mode, name in iterable:
            path = join_path(self.path, name)
            try:
                yield self._map_id_to_type[mode >> 12](self.repo, binsha, mode, path)
            except KeyError:
                raise TypeError("Unknown mode %o found in tree data for path '%s'" % (mode, path))
534490
def _to_java(self):
        

        estimator, epms, evaluator = super(TrainValidationSplit, self)._to_java_impl()

        _java_obj = JavaParams._new_java_obj("org.apache.spark.ml.tuning.TrainValidationSplit",
                                             self.uid)
        _java_obj.setEstimatorParamMaps(epms)
        _java_obj.setEvaluator(evaluator)
        _java_obj.setEstimator(estimator)
        _java_obj.setTrainRatio(self.getTrainRatio())
        _java_obj.setSeed(self.getSeed())
        _java_obj.setParallelism(self.getParallelism())
        _java_obj.setCollectSubModels(self.getCollectSubModels())
        return _java_obj
586756
@SuppressWarnings("unchecked") // always have unchecked casts when dealing with raw classes
    private Geometry decodeObject(final List mongoDBGeometry, final List<GeometryFactory> geometryFactories) {
        GeometryFactory factory = geometryFactories.get(0);
        if (geometryFactories.size() == 1) {
            // This should be the last list, so no need to decode further
            return factory.createGeometry(mongoDBGeometry);
        } else {
            List<Geometry> decodedObjects = new ArrayList<Geometry>();
            for (final Object objectThatNeedsDecoding : mongoDBGeometry) {
                // MongoDB geometries are lists of lists of lists...
                decodedObjects.add(decodeObject((List) objectThatNeedsDecoding,
                                                geometryFactories.subList(1, geometryFactories.size())));
            }
            return factory.createGeometry(decodedObjects);
        }
    }
807093
def callMLlibFunc(name, *args):
    
    sc = SparkContext.getOrCreate()
    api = getattr(sc._jvm.PythonMLLibAPI(), name)
    return callJavaFunc(sc, api, *args)
309849
public DataFrame createJavaSchemaRDD(ExtractorConfig<Cells> config) throws UnsupportedDataTypeException, UnsupportedOperationException {
        JavaRDD<Cells> cellsRDD = createJavaRDD(config);
        JavaRDD<Row> rowsRDD = DeepSparkContext.createJavaRowRDD(cellsRDD);
        try {
            Cells firstCells = cellsRDD.first();
            StructType schema = CellsUtils.getStructTypeFromCells(firstCells);
            return sqlContext.applySchema(rowsRDD, schema);
        } catch(UnsupportedOperationException e) {
            throw new UnsupportedOperationException("Cannot infer schema from empty data RDD", e);
        }
    }