code: def _from_spark_xshards_ray_api(
      spark_xshards: "SparkXShards",
  ) -> "RayXShards":
      ray_ctx = OrcaRayContext.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}"
          store = (
              ray.remote(num_cpus=0, resources={node: 1e-4})(LocalStore)
              .options(name=name, lifetime="detached")
              .remote()
          )
          partition_stores[name] = store
  
      # actor creation is async, this is to make sure they all have been started
      ray.get(
          [
              v.get_partitions_refs.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()
  
      return RayXShards(uuid_str, result_rdd, partition_stores)
  
imports: from __future__ import annotations
  from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from bigdl.orca.data import XShards
  from bigdl.orca.ray import OrcaRayContext
  import logging
  from bigdl.dllib.utils.log4Error import invalidInputError
  from numpy import ndarray
  from typing import TYPE_CHECKING, Any, Callable, DefaultDict, Dict, List, Tuple, Union, no_type_check
label_desc: Use image node_id 16 for calling the RayXShards obj's underlying member method code with example usage: obj._from_spark_xshards_ray_api(spark_xshards) and returns: RayXShards