code: def train_func(
      envs,
      model_dir,
      ds_graph,
      elem_spec,
      val_ds_graph,
      val_elem_sepc,
      fit_kwargs,
  ):
      import tensorflow as tf
      from tensorflow.python.distribute.coordinator.values import (
          deserialize_dataset_from_graph,
      )
  
      import horovod.tensorflow.keras as hvd
  
      os.environ.update(envs[hvd.rank()])
  
      tf.config.threading.set_inter_op_parallelism_threads(1)
      tf.config.threading.set_intra_op_parallelism_threads(
          int(os.environ["OMP_NUM_THREADS"])
      )
  
      import horovod.tensorflow.keras as hvd
  
      new_model = hvd.load_model(os.path.join(model_dir, "temp_model"))
      train_dataset = deserialize_dataset_from_graph(
          ds_graph, elem_spec
      )
      if val_ds_graph is not None:
          val_dataset = deserialize_dataset_from_graph(
              val_ds_graph, val_elem_sepc
          )
      else:
          val_dataset = None
  
      from tensorflow.python.distribute.input_ops import (
          auto_shard_dataset,
      )
  
      train_dataset = rebatch_dataset(
          train_dataset, hvd.size(), hvd.rank()
      )
      train_dataset = auto_shard_dataset(
          train_dataset, hvd.size(), hvd.rank(), hvd.size()
      )
  
      if val_dataset is not None:
          val_dataset = rebatch_dataset(
              val_dataset, hvd.size(), hvd.rank()
          )
          val_dataset = auto_shard_dataset(
              val_dataset, hvd.size(), hvd.rank(), hvd.size()
          )
  
      task_id = hvd.rank()
  
      if task_id == 0:
          verbose = fit_kwargs["verbose"]
      else:
          verbose = 0
      del fit_kwargs["verbose"]
      history = new_model.fit(
          train_dataset,
          validation_data=val_dataset,
          verbose=verbose,
          **fit_kwargs,
      )
      if task_id == 0:
          path = os.path.join(model_dir, "trained_model_weights")
          new_model.save_weights(path, overwrite=True)
      else:
          path = os.path.join(
              model_dir, f"trained_model_weights_{task_id}"
          )
          new_model.save_weights(path, overwrite=True)
      return history
  
file_path: BigDL/python/nano/src/bigdl/nano/deps/horovod/distributed_utils_horovod.py
imports: import os
  import json
  import shutil
  from tempfile import TemporaryDirectory
  from contextlib import closing
  import socket
  import tensorflow
  from bigdl.nano.utils.common import schedule_processors
  from bigdl.nano.utils.common import invalidInputError
label_desc: Use image node_id 3 for calling a global function with example usage: train_func(envs, model_dir, ds_graph, elem_spec, val_ds_graph, val_elem_sepc, fit_kwargs) and returns: history