code: def __init__(
      self,
      model_creator: Optional[Callable] = None,
      config: Optional[Dict] = None,
      compile_args_creator: Optional[Callable] = None,
      verbose: bool = False,
      workers_per_node: int = 1,
      model_dir: Optional[str] = None,
      log_to_driver: bool = True,
      **kwargs,
  ) -> None:
      self.model_creator = model_creator
      self.compile_args_creator = compile_args_creator
      self.config = {} if config is None else config
      self.verbose = verbose
  
      sc = OrcaContext.get_spark_context()
  
      num_node, num_core = get_node_and_core_number()
      self.num_workers = num_node * workers_per_node
      self.total_cores = num_node * num_core
  
      # over partition to cover tasks all over the cluster
      self.workerRDD = sc.parallelize(
          list(range(self.total_cores * 4)), self.total_cores * 4
      ).repartition(self.num_workers)
  
      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"] = (
              num_core // workers_per_node
          )
  
      self.model_weights = None
      self.optimizer_weights = None
      self.load_path = None
      self.load_params = None
  
      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.",
          )
      self.model_dir = parse_model_dir(model_dir)  # type:ignore
      master = sc.getConf().get("spark.master")
      if not master.startswith("local"):
          logger.info(
              "For cluster mode, make sure to use shared filesystem path "
              "as model directory."
          )
  
      self.application_id = sc.applicationId
      self.ip = get_node_ip()
      self.port = find_free_port()
      is_local = sc.master.startswith("local")
      self.need_to_log_to_driver = (not is_local) and log_to_driver
      if self.need_to_log_to_driver:
          self.log_server_thread = start_log_server(self.ip, self.port)
  
imports: import os
  import types
  import logging
  import tempfile
  import shutil
  from bigdl.dllib.utils.common import get_node_and_core_number
  from bigdl.dllib.utils.file_utils import is_local_path
  from bigdl.dllib.utils.utils import get_node_ip
  from bigdl.orca.data.file import is_file, exists, get_remote_file_to_local, get_remote_files_with_prefix_to_local, put_local_file_to_remote, put_local_files_with_prefix_to_remote
  from bigdl.orca.learn.tf2.spark_runner import SparkRunner
  from bigdl.orca.learn.utils import find_free_port, find_ip_and_free_port
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards, dataframe_to_xshards, convert_predict_xshards_to_dataframe, make_data_creator, load_model, save_model, process_xshards_of_pandas_dataframe, add_predict_to_pd_xshards, update_predict_xshards
  from bigdl.orca.learn.log_monitor import start_log_server, stop_log_server
  from bigdl.orca import OrcaContext
  from bigdl.orca.data.shard import SparkXShards
  from bigdl.dllib.utils.log4Error import invalidInputError
  from pyspark.sql.dataframe import DataFrame
  from typing import TYPE_CHECKING, Any, Dict, List, Callable, Union, Optional
label_desc: Use image node_id 1 to create a new SparkTFEstimator object  with example: obj = SparkTFEstimator(model_creator, config, compile_args_creator, verbose, workers_per_node, model_dir, log_to_driver)