code: def __init__(
      self,
      model_creator: Optional["Callable"] = None,
      compile_args_creator: Optional["Callable"] = None,
      config: Optional[Dict] = None,
      verbose: bool = False,
      backend: str = "ray",
      workers_per_node: int = 1,
      cpu_binding: bool = False,
  ) -> None:
      self.model_creator = model_creator
      self.compile_args_creator = compile_args_creator
      self.config = {} if config is None else config
      self.verbose = verbose
  
      if self.model_creator is None:
          logger.warning(
              "Please use load function of the estimator to load model when"
              " model_creator is None."
          )
  
      ray_ctx = OrcaRayContext.get()
      if "batch_size" in self.config:
          invalidInputError(
              False,
              "Please do not specify batch_size in config. Input batch_size in the"
              " fit/evaluate function of the estimator instead.",
          )
  
      if "inter_op_parallelism" not in self.config:
          self.config["inter_op_parallelism"] = 1
  
      if "intra_op_parallelism" not in self.config:
          ray_node_cpu_cores = ray_ctx.ray_node_cpu_cores  # type:ignore
          self.config["intra_op_parallelism"] = (
              ray_node_cpu_cores // workers_per_node
          )
  
      if backend == "horovod":
          invalidInputError(
              compile_args_creator is not None,
              "compile_args_creator should not be None,"
              " when backend is set to horovod.",
          )
  
      params = {
          "model_creator": self.model_creator,
          "compile_args_creator": self.compile_args_creator,
          "config": self.config,
          "verbose": self.verbose,
      }
  
      if backend == "ray":
          cores_per_node = (
              ray_ctx.ray_node_cpu_cores // workers_per_node
          )  # type:ignore
          num_nodes = (
              ray_ctx.num_ray_nodes * workers_per_node
          )  # type:ignore
  
          self.cluster = RayDLCluster(
              num_workers=num_nodes,
              worker_cores=cores_per_node,
              worker_cls=TFRunner,
              worker_param=params,
              cpu_binding=cpu_binding,
          )
          self.remote_workers = self.cluster.get_workers()
          ips = ray.get(
              [
                  worker.get_node_ip.remote()
                  for worker in self.remote_workers
              ]
          )
          ports = ray.get(
              [
                  worker.find_free_port.remote()
                  for worker in self.remote_workers
              ]
          )
  
          urls = [
              "{ip}:{port}".format(ip=ips[i], port=ports[i])
              for i in range(len(self.remote_workers))
          ]
          ray.get(
              [worker.setup.remote() for worker in self.remote_workers]
          )
          # Get setup tasks in order to throw errors on failure
          ray.get(
              [
                  worker.setup_distributed.remote(
                      urls, i, len(self.remote_workers)
                  )
                  for i, worker in enumerate(self.remote_workers)
              ]
          )
      elif backend == "horovod":
          # it is necessary to call self.run first to set horovod environment
          from bigdl.orca.learn.horovod.horovod_ray_runner import (
              HorovodRayRunner,
          )
  
          horovod_runner = HorovodRayRunner(
              ray_ctx,
              worker_cls=TFRunner,
              worker_param=params,
              workers_per_node=workers_per_node,
          )
          horovod_runner.run(lambda: print("worker initialized"))
          self.remote_workers = horovod_runner.remote_workers
          ray.get(
              [worker.setup.remote() for worker in self.remote_workers]
          )
          ray.get(
              [
                  worker.setup_horovod.remote()
                  for i, worker in enumerate(self.remote_workers)
              ]
          )
      else:
          invalidInputError(
              False,
              'Only "ray" and "horovod" are legal '
              "values of backend, but got {}".format(backend),
          )
  
      self.num_workers = len(self.remote_workers)
  
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 1 to create a new TensorFlow2Estimator object from inherited base classes: OrcaRayEstimator with example: obj = TensorFlow2Estimator(model_creator, compile_args_creator, config, verbose, backend, workers_per_node, cpu_binding)