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 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.
  :param label_cols: label column names if train data is Spark DataFrame.
  :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","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",".",":","param","label_cols",":","label","column","names","if","train","data","is","Spark","DataFrame",".",":","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",
          Callable[[Dict, int], "DataLoader"],
      ],
      batch_size: int = 32,
      num_steps: Optional[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 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.
      :param label_cols: label column names if train data is Spark DataFrame.
      :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
  
      if self.model_creator is None:
          invalidInputError(
              False,
              "Must provide callable function for model_creator "
              "or load a saved model.",
          )
  
      sc = OrcaContext.get_spark_context()
      state_dict = self._get_broadcasted_state_dict(sc)
      init_params = dict(mode="evaluate", state_dict=state_dict)
      init_params.update(self.worker_init_params)
  
      # 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,
          callbacks=callbacks,
      )
  
      if isinstance(data, SparkXShards):
          data = data.to_lazy()
      if isinstance(data, DataFrame) or isinstance(data, SparkXShards):
          if data.rdd.getNumPartitions() != self.num_workers:
              data = data.repartition(self.num_workers)
      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,
      )
      if isinstance(
          data, SparkXShards
      ):  # Computation triggered when collect
          params["wrap_dataloader"] = False
          if data._get_class_name() == "pandas.core.frame.DataFrame":
              data = process_xshards_of_pandas_dataframe(
                  data, feature_cols, label_cols
              )
  
          def transform_func(iter, init_param, param):
              partition_data = list(iter)
              param["data_creator"] = partition_to_creator(
                  partition_data
              )
              return PytorchPysparkWorker(**init_param).validate(
                  **param
              )
  
          data_rdd = data.rdd  # type:ignore
          res = (
              data_rdd.barrier()
              .mapPartitions(
                  lambda iter: transform_func(iter, init_params, params)
              )
              .collect()
          )
      else:
          params["data_creator"] = data  # type:ignore
  
          def transform_func(iter, init_param, param):
              return PytorchPysparkWorker(**init_param).validate(
                  **param
              )
  
          res = (
              self.workerRDD.barrier()
              .mapPartitions(
                  lambda iter: transform_func(iter, init_params, params)
              )
              .collect()
          )
  
      if reduce_results:
          return process_stats(res)
      else:
          return res
  
imports: import types
  import torch
  import copy
  import os
  import shutil
  import tempfile
  import logging
  import math
  from pyspark.sql.dataframe import DataFrame
  from bigdl.dllib.utils.file_utils import is_local_path
  from bigdl.orca.learn.pytorch.pytorch_pyspark_worker import PytorchPysparkWorker
  from bigdl.orca.learn.pytorch.utils import process_stats
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards, dataframe_to_xshards, convert_predict_xshards_to_dataframe, make_data_creator, update_predict_xshards, process_xshards_of_pandas_dataframe, add_predict_to_pd_xshards
  from bigdl.orca.data import SparkXShards
  from bigdl.orca import OrcaContext
  from bigdl.orca.learn.base_estimator import BaseEstimator
  from bigdl.orca.data.file import get_remote_file_to_local, put_local_file_to_remote
  from bigdl.dllib.utils.common import get_node_and_core_number
  from bigdl.orca.learn.log_monitor import start_log_server, stop_log_server
  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
  from bigdl.orca.learn.utils import find_free_port, find_ip_and_free_port
  from bigdl.dllib.utils.utils import get_node_ip
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING, Union, Optional, Callable, Dict, List
  from bigdl.orca.learn.pytorch.callbacks import Callback
label_desc: Use image node_id 9 for calling the PyTorchPySparkEstimator 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, res, PytorchPysparkWorker, PytorchPysparkWorker