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)
          result.append(
              local_store.upload_shards.remote(
                  (idx, shard_id), shard_ref
              )
          )
      is_empty = len(result) == 0
      if is_empty:
          partition_ref = ray.put([])
          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 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 2 for calling a global function with example usage: write_to_ray(idx, partition, redis_address, redis_password, partition_store_names) and returns: list