code: def _fit_ray_xshards(self, train_shards, val_shards, params):
      if val_shards is None:
  
          def transform_func(worker, partition_refs):
              params["data_creator"] = make_data_creator(partition_refs)
              return worker.step.remote(**params)
  
          worker_stats = train_shards.reduce_partitions_for_actors(
              self.remote_workers, transform_func
          )
      else:
  
          def zip_func(
              worker, this_partition_refs, that_partition_refs
          ):
              params["data_creator"] = make_data_creator(
                  this_partition_refs
              )
              params["validation_data_creator"] = make_data_creator(
                  that_partition_refs
              )
              return worker.step.remote(**params)
  
          worker_stats = train_shards.zip_reduce_shards_with_actors(
              val_shards, self.remote_workers, zip_func
          )
      return worker_stats
  
imports: import os
  import types
  from bigdl.orca.common import SafePickle
  import shutil
  import tempfile
  import logging
  import ray
  from bigdl.dllib.utils import log4Error
  from bigdl.dllib.utils.file_utils import is_local_path
  from bigdl.orca.data.file import enable_multi_fs_save, enable_multi_fs_load
  from bigdl.orca.data.ray_xshards import RayXShards
  from bigdl.orca.learn.dl_cluster import RayDLCluster
  from bigdl.orca.learn.tf2.tf_runner import TFRunner
  from bigdl.orca.learn.ray_estimator import Estimator
  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, make_data_creator, add_predict_to_pd_xshards
  from bigdl.orca.data.file import get_remote_file_to_local, get_remote_dir_to_local, is_file
  from bigdl.orca.data.utils import process_spark_xshards
  from bigdl.dllib.utils.log4Error import invalidInputError
  from bigdl.orca.ray import OrcaRayContext
  from typing import TYPE_CHECKING, Dict, List, Callable, Union, Optional
label_desc: Use image node_id 3 for calling the TensorFlow2Estimator obj's underlying member method code with example usage: obj._fit_ray_xshards(train_shards, val_shards, params) and returns: worker_stats, worker, worker