docstr: Evenly allocate partitions for actors and run `reduce_partitions_func` on partitions of each
  worker.
  Return a list of results, where one result corresponds to one worker.
  
  :param actors: ray actors
  :param reduce_partitions_func: Function to run on each ray actor which reduces the
      partition refs on the actor to one result_ref. It should take an actor and a list of
      partition_refs as argument return a result_ref
  :param return_refs: Whether to return ray objects refs or ray objects. If True, return a
  list of ray object refs, otherwise return a list of ray objects. Defaults to be False,
docstr_tok: ["Evenly","allocate","partitions","for","actors","and","run","`","reduce_partitions_func","`","on","partitions","of","each","worker",".","Return","a","list","of","results",",","where","one","result","corresponds","to","one","worker",".",":","param","actors",":","ray","actors",":","param","reduce_partitions_func",":","Function","to","run","on","each","ray","actor","which","reduces","the","partition","refs","on","the","actor","to","one","result_ref",".","It","should","take","an","actor","and","a","list","of","partition_refs","as","argument","return","a","result_ref",":","param","return_refs",":","Whether","to","return","ray","objects","refs","or","ray","objects",".","If","True",",","return","a","list","of","ray","object","refs",",","otherwise","return","a","list","of","ray","objects",".","Defaults","to","be","False",","]
code: def reduce_partitions_for_actors(
      self,
      actors: List["ActorHandle"],
      reduce_partitions_func: Callable,
      return_refs: bool = False,
  ) -> List[Any]:
      """
      Evenly allocate partitions for actors and run `reduce_partitions_func` on partitions of each
      worker.
      Return a list of results, where one result corresponds to one worker.
  
      :param actors: ray actors
      :param reduce_partitions_func: Function to run on each ray actor which reduces the
          partition refs on the actor to one result_ref. It should take an actor and a list of
          partition_refs as argument return a result_ref
      :param return_refs: Whether to return ray objects refs or ray objects. If True, return a
      list of ray object refs, otherwise return a list of ray objects. Defaults to be False,
      """
      invalidInputError(
          self.num_partitions() >= len(actors),
          f"Get the number of partitions ({self.num_partitions()}) smaller than "
          f"the number of workers ({len(actors)}).",
      )
      assigned_partitions, _, _ = self.assign_partitions_to_actors(
          actors
      )
      result_refs = []
      for actor, part_ids in zip(actors, assigned_partitions):
          assigned_partition_refs = self._get_multiple_partition_refs(
              part_ids
          )
          result_ref = reduce_partitions_func(
              actor, assigned_partition_refs
          )
          result_refs.append(result_ref)
      if return_refs:
          return result_refs
      results = ray.get(result_refs)
      return results
  
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 11 for calling the RayXShards obj's underlying member method code with example usage: obj.reduce_partitions_for_actors(actors, reduce_partitions_func, return_refs) and returns: results, result_refs