docstr: Evaluates a PyTorch model given validation data.
  Note that only accuracy for classification with zero-based label is supported by
  default. You can override validate_batch in TorchRunner for other metrics.
  Calls `TorchRunner.validate()` on N parallel workers simultaneously
  underneath the hood.
  
  :param data: An instance of SparkXShards, a Spark DataFrame, a Ray Dataset or a function
         that takes config and batch_size as argument and returns a PyTorch DataLoader for
         validation.
  :param batch_size: Total batch size for all workers used for evaluation. Each worker's batch
         size would be this value divide the total number of workers. Default: 32.
         If your validation data is a function, you can set batch_size to be the input
         batch_size of the function for the PyTorch DataLoader.
  :param num_steps: The number of batches to compute the validation results on. This
         corresponds to the number of times `TorchRunner.validate_batch` is called.
  :param profile: Boolean. Whether to return time stats for the training procedure.
         Default is False.
  :param reduce_results: Boolean. Whether to average all metrics across all workers into
         one dict. If a metric is a non-numerical value, the one value will be randomly
         selected among the workers. If False, returns a list of dicts for
         all workers. Default is True.
  :param feature_cols: feature column names if train data is Spark DataFrame or Ray Dataset.
  :param label_cols: label column names if train data is Spark DataFrame or Ray Dataset.
  :param callbacks: A list for all callbacks. Note that only one MainCallback
         is allowed among all callbacks.
  
  :return: A dictionary of metrics for the given data, including validation accuracy and loss.
          You can also provide custom metrics by passing in a custom HookClass(after 2.2.0)
          when creating the Estimator.
docstr_tok: ["Evaluates","a","PyTorch","model","given","validation","data",".","Note","that","only","accuracy","for","classification","with","zero-based","label","is","supported","by","default",".","You","can","override","validate_batch","in","TorchRunner","for","other","metrics",".","Calls","`","TorchRunner.validate","(",")","`","on","N","parallel","workers","simultaneously","underneath","the","hood",".",":","param","data",":","An","instance","of","SparkXShards",",","a","Spark","DataFrame",",","a","Ray","Dataset","or","a","function","that","takes","config","and","batch_size","as","argument","and","returns","a","PyTorch","DataLoader","for","validation",".",":","param","batch_size",":","Total","batch","size","for","all","workers","used","for","evaluation",".","Each","worker","'s","batch","size","would","be","this","value","divide","the","total","number","of","workers",".","Default",":","32",".","If","your","validation","data","is","a","function",",","you","can","set","batch_size","to","be","the","input","batch_size","of","the","function","for","the","PyTorch","DataLoader",".",":","param","num_steps",":","The","number","of","batches","to","compute","the","validation","results","on",".","This","corresponds","to","the","number","of","times","`","TorchRunner.validate_batch","`","is","called",".",":","param","profile",":","Boolean",".","Whether","to","return","time","stats","for","the","training","procedure",".","Default","is","False",".",":","param","reduce_results",":","Boolean",".","Whether","to","average","all","metrics","across","all","workers","into","one","dict",".","If","a","metric","is","a","non-numerical","value",",","the","one","value","will","be","randomly","selected","among","the","workers",".","If","False",",","returns","a","list","of","dicts","for","all","workers",".","Default","is","True",".",":","param","feature_cols",":","feature","column","names","if","train","data","is","Spark","DataFrame","or","Ray","Dataset",".",":","param","label_cols",":","label","column","names","if","train","data","is","Spark","DataFrame","or","Ray","Dataset",".",":","param","callbacks",":","A","list","for","all","callbacks",".","Note","that","only","one","MainCallback","is","allowed","among","all","callbacks",".",":","return",":","A","dictionary","of","metrics","for","the","given","data",",","including","validation","accuracy","and","loss",".","You","can","also","provide","custom","metrics","by","passing","in","a","custom","HookClass","(","after","2.2.0",")","when","creating","the","Estimator","."]
code: def evaluate(
      self,
      data: Union[
          "SparkXShards",
          "SparkDataFrame",
          "RayDataset",
          Callable[[Dict, int], "DataLoader"],
      ],
      batch_size: int = 32,
      num_steps: int = None,
      profile: bool = False,
      reduce_results: bool = True,
      feature_cols: Optional[List[str]] = None,
      label_cols: Optional[List[str]] = None,
      callbacks: Optional[List["Callback"]] = None,
  ) -> Union[List[Dict], Dict]:
      """
      Evaluates a PyTorch model given validation data.
      Note that only accuracy for classification with zero-based label is supported by
      default. You can override validate_batch in TorchRunner for other metrics.
      Calls `TorchRunner.validate()` on N parallel workers simultaneously
      underneath the hood.
  
      :param data: An instance of SparkXShards, a Spark DataFrame, a Ray Dataset or a function
             that takes config and batch_size as argument and returns a PyTorch DataLoader for
             validation.
      :param batch_size: Total batch size for all workers used for evaluation. Each worker's batch
             size would be this value divide the total number of workers. Default: 32.
             If your validation data is a function, you can set batch_size to be the input
             batch_size of the function for the PyTorch DataLoader.
      :param num_steps: The number of batches to compute the validation results on. This
             corresponds to the number of times `TorchRunner.validate_batch` is called.
      :param profile: Boolean. Whether to return time stats for the training procedure.
             Default is False.
      :param reduce_results: Boolean. Whether to average all metrics across all workers into
             one dict. If a metric is a non-numerical value, the one value will be randomly
             selected among the workers. If False, returns a list of dicts for
             all workers. Default is True.
      :param feature_cols: feature column names if train data is Spark DataFrame or Ray Dataset.
      :param label_cols: label column names if train data is Spark DataFrame or Ray Dataset.
      :param callbacks: A list for all callbacks. Note that only one MainCallback
             is allowed among all callbacks.
  
      :return: A dictionary of metrics for the given data, including validation accuracy and loss.
              You can also provide custom metrics by passing in a custom HookClass(after 2.2.0)
              when creating the Estimator.
      """
      invalidInputError(
          isinstance(batch_size, int) and batch_size > 0,
          "batch_size should be a positive integer",
      )
      batch_size = (
          batch_size // self.num_workers
      )  # Local batch size for each worker
      if batch_size <= 0:
          batch_size = 1
      from bigdl.orca.data import SparkXShards
  
      data, _ = maybe_dataframe_to_xshards(
          data,
          validation_data=None,
          feature_cols=feature_cols,
          label_cols=label_cols,
          mode="evaluate",
          num_workers=self.num_workers,
          shard_size=batch_size,
      )
  
      # Check uniqueness of the MainCallback
      callbacks = callbacks or []
      make_only_mainCallback(callbacks)
      if self.use_tqdm and not is_tqdm_exists(callbacks):
          callbacks.append(TqdmCallback())
  
      params = dict(
          batch_size=batch_size,
          num_steps=num_steps,
          profile=profile,
          wrap_dataloader=False,
          callbacks=callbacks,
      )
  
      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 bigdl.orca.data.utils import process_spark_xshards
  
          ray_xshards = process_spark_xshards(
              data, self.num_workers
          )  # type:ignore
  
          def transform_func(worker, partition_refs):
              data_creator = partition_refs_to_creator(partition_refs)
              # Should not wrap DistributedSampler on DataLoader for SparkXShards input.
              params["data_creator"] = data_creator
              return worker.validate.remote(**params)
  
          worker_stats = ray_xshards.reduce_partitions_for_actors(
              self.remote_workers, transform_func
          )
      elif isinstance(data, ray.data.Dataset):
          shards = data.split(
              n=self.num_workers, locality_hints=self.remote_workers
          )
  
          def data_creator(config, batch_size):
              torch_datashard = shard.to_torch(
                  label_column=label_cols,
                  feature_columns=feature_cols,
                  batch_size=batch_size,
              )
              return torch_datashard
  
          remote_worker_stats = []
          params["data_creator"] = data_creator  # type:ignore
          for shard, worker in zip(shards, self.remote_workers):
              stats = worker.validate.remote(**params)
              remote_worker_stats.append(stats)
          worker_stats = ray.get(remote_worker_stats)
      else:
          invalidInputError(
              isinstance(data, types.FunctionType),
              "data should be either an instance of SparkXShards or a callable"
              " function, but got type: {}".format(type(data)),
          )
  
          params["data_creator"] = data  # type:ignore
          worker_stats = ray.get(
              [w.validate.remote(**params) for w in self.remote_workers]
          )
  
      if reduce_results:
          return process_stats(worker_stats)
      else:
          return worker_stats
  
imports: import types
  import copy
  import logging
  import math
  from bigdl.orca.data.ray_xshards import RayXShards
  from bigdl.orca.learn.pytorch.pytorch_ray_worker import PytorchRayWorker
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards, dataframe_to_xshards, convert_predict_xshards_to_dataframe, update_predict_xshards, process_xshards_of_pandas_dataframe, add_predict_to_pd_xshards
  from bigdl.orca.ray import OrcaRayContext
  from bigdl.orca.learn.pytorch.core.base_ray_estimator import BaseRayEstimator
  from bigdl.orca.learn.pytorch.utils import process_stats, check_for_failure
  from bigdl.orca.learn.pytorch.callbacks.maincallback import make_only_mainCallback
  from bigdl.orca.learn.pytorch.callbacks.tqdm import TqdmCallback, is_tqdm_exists
  from bigdl.orca.learn.pytorch.callbacks.maxsteps import MaxstepsCallback
  import ray
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING, Union, Optional, Callable, Dict, List
label_desc: Use image node_id 4 for calling the PyTorchRayEstimator obj's underlying member method code with example usage: obj.evaluate(data, batch_size, num_steps, profile, reduce_results, feature_cols, label_cols, callbacks) and returns: process_stats, worker_stats, worker, torch_datashard