docstr: Assign each partition_ref (referencing a list of shards) to an actor,
  and run func for each actor and partition_ref pair.
  Actors should have a `get_node_ip` method to achieve locality scheduling.
  The `get_node_ip` method should call ray._private.services.get_node_ip_address()
  to return the correct ip address.
  The `func` should take an actor and a partition_ref as argument and
  invoke some remote func on that actor and return a new partition_ref.
  Note that if you pass partition_ref directly to actor method, ray
  will resolve that partition_ref to the actual partition object, which
  is a list of shards. If you pass partition_ref indirectly through other
  object, say [partition_ref], ray will send the partition_ref itself to
  actor, and you may need to use ray.get(partition_ref) on actor to retrieve
  the actor partition objects.
docstr_tok: ["Assign","each","partition_ref","(","referencing","a","list","of","shards",")","to","an","actor",",","and","run","func","for","each","actor","and","partition_ref","pair",".","Actors","should","have","a","`","get_node_ip","`","method","to","achieve","locality","scheduling",".","The","`","get_node_ip","`","method","should","call","ray._private.services.get_node_ip_address","(",")","to","return","the","correct","ip","address",".","The","`","func","`","should","take","an","actor","and","a","partition_ref","as","argument","and","invoke","some","remote","func","on","that","actor","and","return","a","new","partition_ref",".","Note","that","if","you","pass","partition_ref","directly","to","actor","method",",","ray","will","resolve","that","partition_ref","to","the","actual","partition","object",",","which","is","a","list","of","shards",".","If","you","pass","partition_ref","indirectly","through","other","object",",","say","[","partition_ref","]",",","ray","will","send","the","partition_ref","itself","to","actor",",","and","you","may","need","to","use","ray.get","(","partition_ref",")","on","actor","to","retrieve","the","actor","partition","objects","."]
code: def transform_shards_with_actors(
      self, actors: List["ActorHandle"], func: Callable
  ) -> "RayXShards":
      """
      Assign each partition_ref (referencing a list of shards) to an actor,
      and run func for each actor and partition_ref pair.
      Actors should have a `get_node_ip` method to achieve locality scheduling.
      The `get_node_ip` method should call ray._private.services.get_node_ip_address()
      to return the correct ip address.
      The `func` should take an actor and a partition_ref as argument and
      invoke some remote func on that actor and return a new partition_ref.
      Note that if you pass partition_ref directly to actor method, ray
      will resolve that partition_ref to the actual partition object, which
      is a list of shards. If you pass partition_ref indirectly through other
      object, say [partition_ref], ray will send the partition_ref itself to
      actor, and you may need to use ray.get(partition_ref) on actor to retrieve
      the actor partition objects.
      """
      (
          assigned_partitions,
          actor_ips,
          assigned_actors,
      ) = self.assign_partitions_to_actors(actors)
      assigned_partition_refs = [
          (part_ids, self._get_multiple_partition_refs(part_ids))
          for part_ids in assigned_partitions
      ]
      new_part_id_refs = {
          part_id: func(actor, part_ref)
          for actor, (part_ids, part_refs) in zip(
              assigned_actors, assigned_partition_refs
          )
          for part_id, part_ref in zip(part_ids, part_refs)
      }
  
      actor_ip2part_id = defaultdict(list)
      for actor_ip, part_ids in zip(actor_ips, assigned_partitions):
          actor_ip2part_id[actor_ip].extend(part_ids)
  
      return RayXShards.from_partition_refs(
          actor_ip2part_id, new_part_id_refs, self.rdd
      )
  
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 10 for calling the RayXShards obj's underlying member method code with example usage: obj.transform_shards_with_actors(actors, func) and returns: RayXShards