code: def _from_spark_xshards_ray_api(spark_xshards):
      ray_ctx = RayContext.get()
      address = ray_ctx.redis_address
      password = ray_ctx.redis_password
      driver_ip = ray._private.services.get_node_ip_address()
      uuid_str = str(uuid.uuid4())
      resources = ray.cluster_resources()
      nodes = []
      for key, value in resources.items():
          if key.startswith("node:"):
              # if running in cluster, filter out driver ip
              if key != f"node:{driver_ip}":
                  nodes.append(key)
      # for the case of local mode and single node spark standalone
      if not nodes:
          nodes.append(f"node:{driver_ip}")
  
      partition_stores = {}
      for node in nodes:
          name = f"partition:{uuid_str}:{node}"
          if version.parse(ray.__version__) >= version.parse("1.4.0"):
              store = (
                  ray.remote(num_cpus=0, resources={node: 1e-4})(
                      LocalStore
                  )
                  .options(name=name, lifetime="detached")
                  .remote()
              )
          else:
              store = (
                  ray.remote(num_cpus=0, resources={node: 1e-4})(
                      LocalStore
                  )
                  .options(name=name)
                  .remote()
              )
          partition_stores[name] = store
  
      # actor creation is aync, this is to make sure they all have been started
      ray.get(
          [v.get_partitions.remote() for v in partition_stores.values()]
      )
      partition_store_names = list(partition_stores.keys())
      result_rdd = spark_xshards.rdd.mapPartitionsWithIndex(
          lambda idx, part: write_to_ray(
              idx, part, address, password, partition_store_names
          )
      ).cache()
      result = result_rdd.collect()
  
      id2ip = {}
      id2store_name = {}
      for idx, ip, local_store_name in result:
          id2ip[idx] = ip
          id2store_name[idx] = local_store_name
  
      return RayXShards(uuid_str, result_rdd, partition_stores)
  
imports: from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from packaging import version
  from zoo.orca.data import XShards
  from zoo.ray import RayContext
  import logging
label_desc: Use image node_id 14 for calling the RayXShards obj's underlying member method code with example usage: obj._from_spark_xshards_ray_api(spark_xshards) and returns: RayXShards