code: def __init__(
      self,
      model_creator,
      compile_args_creator=None,
      config=None,
      verbose=False,
      backend="tf2",
      workers_per_node=1,
      cpu_binding=False,
  ):
      self.model_creator = model_creator
      self.compile_args_creator = compile_args_creator
      self.config = {} if config is None else config
      self.verbose = verbose
  
      ray_ctx = RayContext.get()
      if "batch_size" in self.config:
          raise Exception(
              "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:
          self.config["intra_op_parallelism"] = (
              ray_ctx.ray_node_cpu_cores // workers_per_node
          )
  
      if backend == "horovod":
          assert compile_args_creator is not None, (
              "compile_args_creator should not be None,"
              " when backend is set to horovod"
          )
  
      params = {
          "model_creator": model_creator,
          "compile_args_creator": compile_args_creator,
          "config": self.config,
          "verbose": self.verbose,
      }
  
      if backend == "tf2":
          cores_per_node = (
              ray_ctx.ray_node_cpu_cores // workers_per_node
          )
          num_nodes = ray_ctx.num_ray_nodes * workers_per_node
  
          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 zoo.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:
          raise Exception(
              'Only "tf2" and "horovod" are legal '
              "values of backend, but got {}".format(backend)
          )
  
      self.num_workers = len(self.remote_workers)
  
imports: import itertools
  import logging
  import pickle
  import numpy
  import ray
  from zoo.common.utils import enable_multi_fs_load, enable_multi_fs_save
  from zoo.orca.data.ray_xshards import RayXShards
  from zoo.orca.learn.dl_cluster import RayDLCluster
  from zoo.orca.learn.tf2.tf_runner import TFRunner
  from zoo.orca.learn.ray_estimator import Estimator
  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.orca.data.utils import process_spark_xshards
  from zoo.ray import RayContext
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)