code: def partition_refs_to_creator(partition_refs):
      def data_creator(config, batch_size):
          from zoo.orca.data.utils import (
              ray_partitions_get_data_label,
              index_data,
              get_size,
          )
          from torch.utils.data import Dataset, DataLoader
  
          class NDArrayDataset(Dataset):
              def __init__(self, x, y):
                  self.x = x  # features
                  self.y = y  # labels
  
              def __len__(self):
                  return get_size(self.y)
  
              def __getitem__(self, i):
                  return index_data(self.x, i), index_data(self.y, i)
  
          params = {"batch_size": batch_size, "shuffle": True}
          for arg in [
              "shuffle",
              "sampler",
              "batch_sampler",
              "num_workers",
              "collate_fn",
              "pin_memory",
              "drop_last",
              "timeout",
              "worker_init_fn",
              "multiprocessing_context",
          ]:
              if arg in config:
                  params[arg] = config[arg]
          data, label = ray_partitions_get_data_label(
              ray.get(partition_refs),
              allow_tuple=False,
              allow_list=False,
          )
          print("Data size on worker: ", len(label))
          dataset = NDArrayDataset(data, label)
          data_loader = DataLoader(dataset, **params)
          return data_loader
  
      return data_creator
  
imports: import io
  import types
  import logging
  import numbers
  import torch
  import numpy
  from zoo.orca.data.ray_xshards import RayXShards
  from zoo.orca.learn.pytorch.training_operator import TrainingOperator
  from zoo.orca.learn.pytorch.torch_runner import TorchRunner
  from zoo.orca.learn.utils import maybe_dataframe_to_xshards, dataframe_to_xshards, convert_predict_xshards_to_dataframe, update_predict_xshards, process_xshards_of_pandas_dataframe
  from zoo.ray import RayContext
  import ray
  from ray.exceptions import RayActorError
label_desc: Use image node_id 2 for calling a global function with example usage: partition_refs_to_creator(partition_refs) and returns: data_creator, data_loader, get_size, index_data,  index_data