code: def _java2py(
      sc: SparkContext,
      r: "JavaObjectOrPickleDump",
      encoding: str = "bytes",
  ) -> Any:
      if isinstance(r, JavaObject):
          clsName = r.getClass().getSimpleName()
          # convert RDD into JavaRDD
          if clsName != "JavaRDD" and clsName.endswith("RDD"):
              r = r.toJavaRDD()
              clsName = "JavaRDD"
  
          assert sc._jvm is not None
  
          if clsName == "JavaRDD":
              jrdd = sc._jvm.org.apache.spark.mllib.api.python.SerDe.javaToPython(
                  r
              )
              return RDD(jrdd, sc)
  
          if clsName == "Dataset":
              return DataFrame(
                  r, SparkSession._getActiveSessionOrCreate()
              )
  
          if clsName in _picklable_classes:
              r = sc._jvm.org.apache.spark.mllib.api.python.SerDe.dumps(
                  r
              )
          elif isinstance(r, (JavaArray, JavaList)):
              try:
                  r = sc._jvm.org.apache.spark.mllib.api.python.SerDe.dumps(
                      r
                  )
              except Py4JJavaError:
                  pass  # not pickable
  
      if isinstance(r, (bytearray, bytes)):
          r = CPickleSerializer().loads(bytes(r), encoding=encoding)
      return r
  
file_path: spark/python/pyspark/mllib/common.py
imports: from typing import Any, Callable, TYPE_CHECKING
  import py4j.protocol
  from py4j.protocol import Py4JJavaError
  from py4j.java_gateway import JavaObject
  from py4j.java_collections import JavaArray, JavaList
  import pyspark.context
  from pyspark import RDD, SparkContext
  from pyspark.serializers import CPickleSerializer, AutoBatchedSerializer
  from pyspark.sql import DataFrame, SparkSession
label_desc: Use image node_id 4 for calling a global function with example usage: _java2py(sc, r, encoding) and returns: r, RDD, DataFrame