code: def write_to_ray(
      idx,
      partition,
      redis_address,
      redis_password,
      partition_store_names,
  ):
      init_ray_if_not(redis_address, redis_password)
      ip = ray._private.services.get_node_ip_address()
      local_store_name = None
      for name in partition_store_names:
          if name.endswith(ip):
              local_store_name = name
              break
      if local_store_name is None:
          local_store_name = random.choice(partition_store_names)
  
      local_store = ray.get_actor(local_store_name)
  
      # directly calling ray.put will set this driver as the owner of this object,
      # when the spark job finished, the driver might exit and make the object
      # eligible for deletion.
      result = []
      for shard_id, shard in enumerate(partition):
          shard_ref = ray.put(shard, _owner=local_store)
          result.append(
              local_store.upload_shards.remote(
                  (idx, shard_id), [shard_ref]
              )
          )
  
      is_empty = len(result) == 0
      if is_empty:
          partition_ref = ray.put([], _owner=local_store)
          result.append(
              local_store.upload_partition.remote(idx, [partition_ref])
          )
          logger.warning(f"Partition {idx} is empty.")
      ray.get(result)
  
      return [(idx, local_store_name.split(":")[-1], local_store_name)]
  
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 2 for calling a global function with example usage: write_to_ray(idx, partition, redis_address, redis_password, partition_store_names) and returns: list