docstr: Initializes the runner.
  Args:
      model_creator (dict -> Model): see tf_trainer.py.
      data_creator (dict -> tf.Dataset, tf.Dataset): see tf_trainer.py.
      config (dict): see tf_trainer.py.
      verbose (bool): Outputs training data if true.
docstr_tok: ["Initializes","the","runner",".","Args",":","model_creator","(","dict","-",">","Model",")",":","see","tf_trainer.py",".","data_creator","(","dict","-",">","tf.Dataset",",","tf.Dataset",")",":","see","tf_trainer.py",".","config","(","dict",")",":","see","tf_trainer.py",".","verbose","(","bool",")",":","Outputs","training","data","if","true","."]
code: def __init__(
      self,
      model_creator=None,
      model_load=None,
      compile_args_creator=None,
      size=None,
      cluster_info=None,
      config=None,
      verbose=False,
      model_weights=None,
      optimizer_weights=None,
      backend="tf-distributed",
      mode="fit",
      model_dir=None,
      application_id=None,
      need_to_log_to_driver=False,
      driver_ip=None,
      driver_port=None,
      filepath=None,
      custom_objects=None,
      compile=True,
  ):
      """Initializes the runner.
      Args:
          model_creator (dict -> Model): see tf_trainer.py.
          data_creator (dict -> tf.Dataset, tf.Dataset): see tf_trainer.py.
          config (dict): see tf_trainer.py.
          verbose (bool): Outputs training data if true.
      """
  
      self.model_creator = model_creator
      self.model_load = model_load
      self.compile_args_creator = compile_args_creator
      self.config = {} if config is None else config
      self.inter_op_parallelism = self.config.get(
          "inter_op_parallelism", 1
      )
      self.intra_op_parallelism = self.config.get(
          "intra_op_parallelism", 1
      )
      self.verbose = verbose
      self.model_weights = model_weights
      self.optimizer_weights = optimizer_weights
      self.size = size
      self.mode = mode
      self.backend = backend
      self.setup()
      self.cluster = cluster_info
      self.custom_objects = custom_objects
      self.compile = compile
      if mode == "fit" or mode == "evaluate":
          from pyspark import BarrierTaskContext
  
          self.partition_id = BarrierTaskContext.get().partitionId()
      else:
          from pyspark import TaskContext
  
          self.partition_id = TaskContext.get().partitionId()
      self.need_to_log_to_driver = need_to_log_to_driver
      if need_to_log_to_driver:
          self.log_path = os.path.join(
              tempfile.gettempdir(),
              "{}_runner.log".format(self.partition_id),
          )
          duplicate_stdout_stderr_to_file(self.log_path)
          (
              self.logger_thread,
              self.thread_stop,
          ) = LogMonitor.start_log_monitor(
              driver_ip=driver_ip,
              driver_port=driver_port,
              log_path=self.log_path,
              partition_id=self.partition_id,
          )
      self.model_dir = model_dir
      self.application_id = application_id
  
      if self.model_creator is None:
          from pyspark import SparkFiles
  
          self.model_load = self.model_load.split("/")[-1]
          self.model_load = SparkFiles.get(self.model_load)
      # create model
      if mode == "fit" or mode == "evaluate":
          if self.backend == "tf-distributed":
              self.setup_distributed(self.cluster)
              if mode == "fit":
                  with self.strategy.scope():
                      if self.model_dir is not None and exists(
                          self._model_saved_path
                      ):
                          # for continous training
                          self.model = load_model(
                              self._model_saved_path
                          )
                      else:
                          if self.model_creator is not None:
                              self.model = self.model_creator(
                                  self.config
                              )
                              if self.optimizer_weights:
  
                                  def build_optimizer():
                                      grad_vars = (
                                          self.model.trainable_weights
                                      )
                                      zero_grads = [
                                          tf.zeros_like(w)
                                          for w in grad_vars
                                      ]
                                      self.model.optimizer.apply_gradients(
                                          zip(zero_grads, grad_vars)
                                      )
                                      self.model.optimizer.set_weights(
                                          self.optimizer_weights.value
                                      )
  
                                  self.strategy.run(build_optimizer)
                              if self.model_weights:
                                  self.model.set_weights(
                                      self.model_weights.value
                                  )
                          else:
                              self.model = tf.keras.models.load_model(
                                  self.model_load,
                                  self.custom_objects,
                                  self.compile,
                              )
  
                      if (
                          not self.model._is_compiled
                          and self.compile_args_creator
                      ):
                          self.model.compile(
                              **self.compile_args_creator(config)
                          )
              else:
                  with self.strategy.scope():
                      if self.model_creator is not None:
                          self.model = self.model_creator(self.config)
                      else:
                          self.model = tf.keras.models.load_model(
                              self.model_load,
                              self.custom_objects,
                              self.compile,
                          )
                      if self.model_weights:
                          self.model.set_weights(
                              self.model_weights.value
                          )
          else:
              self.setup_local()
      else:
          # create local model for predict
          self.setup_local()
  
imports: import json
  import os
  import tempfile
  import shutil
  import copy
  import tensorflow
  from bigdl.orca.data.utils import partition_get_data_label
  from bigdl.orca.data.file import exists
  from bigdl.orca.learn.utils import save_pkl, duplicate_stdout_stderr_to_file, get_rank, process_tensorboard_in_callbacks, save_model, load_model, replace_specific_object_from_callbacks
  from bigdl.orca.learn.log_monitor import LogMonitor
  from bigdl.orca.learn.tf2.callbacks import ModelCheckpoint
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 1 to create a new SparkRunner object  with example: obj = SparkRunner(model_creator, model_load, compile_args_creator, size, cluster_info, config, verbose, model_weights, optimizer_weights, backend, mode, model_dir, application_id, need_to_log_to_driver, driver_ip, driver_port, filepath, custom_objects, compile)