class_docstr: Class responsible for configuring Train dataset preprocessing.
  
  For advanced use cases, this class can be subclassed and the `configure()` method
  overriden for custom data preprocessing.
class_docstr_tok: ["Class","responsible","for","configuring","Train","dataset","preprocessing",".","For","advanced","use","cases",",","this","class","can","be","subclassed","and","the","`","configure","(",")","`","method","overriden","for","custom","data","preprocessing","."]
docstr: Configure how Train datasets should be assigned to workers.
  
  Args:
      datasets: The datasets dict passed to Train by the user.
      world_size: The number of Train workers in total.
      worker_handles: The actor handles of the Train workers.
      worker_node_ids: The node ids of the Train workers.
      kwargs: Forwards compatibility placeholder.
  
  Returns:
      A list of dataset splits for each worker. The size of the list must be
      equal to `world_size`. Each element of the list contains the assigned
      `DataIterator` instances by name for the worker.
docstr_tok: ["Configure","how","Train","datasets","should","be","assigned","to","workers",".","Args",":","datasets",":","The","datasets","dict","passed","to","Train","by","the","user",".","world_size",":","The","number","of","Train","workers","in","total",".","worker_handles",":","The","actor","handles","of","the","Train","workers",".","worker_node_ids",":","The","node","ids","of","the","Train","workers",".","kwargs",":","Forwards","compatibility","placeholder",".","Returns",":","A","list","of","dataset","splits","for","each","worker",".","The","size","of","the","list","must","be","equal","to","`","world_size","`",".","Each","element","of","the","list","contains","the","assigned","`","DataIterator","`","instances","by","name","for","the","worker","."]
code: def configure(
      self,
      datasets: Dict[str, Dataset],
      world_size: int,
      worker_handles: Optional[List[ActorHandle]],
      worker_node_ids: Optional[List[NodeIdStr]],
      **kwargs,
  ) -> List[Dict[str, DataIterator]]:
      """Configure how Train datasets should be assigned to workers.
  
      Args:
          datasets: The datasets dict passed to Train by the user.
          world_size: The number of Train workers in total.
          worker_handles: The actor handles of the Train workers.
          worker_node_ids: The node ids of the Train workers.
          kwargs: Forwards compatibility placeholder.
  
      Returns:
          A list of dataset splits for each worker. The size of the list must be
          equal to `world_size`. Each element of the list contains the assigned
          `DataIterator` instances by name for the worker.
      """
      output = [{} for _ in range(world_size)]
  
      if self._datasets_to_split == "all":
          datasets_to_split = set(datasets.keys())
      else:
          datasets_to_split = set(self._datasets_to_split)
  
      locality_hints = (
          worker_node_ids
          if self._execution_options.locality_with_output
          else None
      )
      for name, ds in datasets.items():
          ds = ds.copy(ds)
          ds.context.execution_options = copy.deepcopy(
              self._execution_options
          )
  
          # Add training-reserved resources to Data's exclude_resources.
          ds.context.execution_options.exclude_resources = (
              ds.context.execution_options.exclude_resources.add(
                  ExecutionResources(
                      cpu=self._num_train_cpus, gpu=self._num_train_gpus
                  )
              )
          )
  
          if name in datasets_to_split:
              for i, split in enumerate(
                  ds.streaming_split(
                      world_size,
                      equal=True,
                      locality_hints=locality_hints,
                  )
              ):
                  output[i][name] = split
          else:
              for i in range(world_size):
                  output[i][name] = ds.iterator()
  
      return output
  
file_path: ray/python/ray/train/_internal/data_config.py
imports: import copy
  from typing import Dict, List, Literal, Optional, Union
  import ray
  from ray.actor import ActorHandle
  from ray.data import DataIterator, Dataset, ExecutionOptions, NodeIdStr
  from ray.data._internal.execution.interfaces.execution_options import ExecutionResources
  from ray.data.preprocessor import Preprocessor
  from ray.train.constants import TRAIN_DATASET_KEY
  from ray.util.annotations import DeveloperAPI, PublicAPI
label_desc: Use image node_id 3 for calling the DataConfig obj's underlying member method code with example usage: obj.configure(datasets, world_size, worker_handles, worker_node_ids) and returns: output