docstr: See the documentation in
  'zoo.orca.learn.pytorch.estimator.PyTorchRayEstimatorWrapper.fit'.
docstr_tok: ["See","the","documentation","in","'zoo.orca.learn.pytorch.estimator.PyTorchRayEstimatorWrapper.fit","'","."]
code: def train(
      self,
      data,
      epochs=1,
      batch_size=32,
      profile=False,
      reduce_results=True,
      info=None,
      feature_cols=None,
      label_cols=None,
  ):
      """
      See the documentation in
      'zoo.orca.learn.pytorch.estimator.PyTorchRayEstimatorWrapper.fit'.
      """
      from zoo.orca.data import SparkXShards
  
      data, _ = maybe_dataframe_to_xshards(
          data,
          validation_data=None,
          feature_cols=feature_cols,
          label_cols=label_cols,
          mode="fit",
          num_workers=self.num_workers,
      )
  
      if isinstance(data, SparkXShards):
          if data._get_class_name() == "pandas.core.frame.DataFrame":
              data = process_xshards_of_pandas_dataframe(
                  data, feature_cols, label_cols
              )
          from zoo.orca.data.utils import process_spark_xshards
  
          ray_xshards = process_spark_xshards(data, self.num_workers)
  
          def transform_func(worker, partition_refs):
              data_creator = partition_refs_to_creator(partition_refs)
              # Should not wrap DistributedSampler on DataLoader for SparkXShards input.
              return worker.train_epochs.remote(
                  data_creator, epochs, batch_size, profile, info, False
              )
  
          worker_stats = ray_xshards.reduce_partitions_for_actors(
              self.remote_workers, transform_func
          )
      else:
          assert isinstance(data, types.FunctionType), (
              "data should be either an instance of SparkXShards or a callable function, but "
              "got type: {}".format(type(data))
          )
  
          success, worker_stats = self._train_epochs(
              data,
              epochs=epochs,
              batch_size=batch_size,
              profile=profile,
              info=info,
          )
  
      epoch_stats = list(map(list, zip(*worker_stats)))
      if reduce_results:
          for i in range(len(epoch_stats)):
              epoch_stats[i] = self._process_stats(epoch_stats[i])
          return epoch_stats
      else:
          return epoch_stats
  
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 the PyTorchRayEstimator obj's underlying member method code with example usage: obj.train(data, epochs, batch_size, profile, reduce_results, info, feature_cols, label_cols) and returns: epoch_stats, epoch_stats, worker