code: def distributed_train_keras(backend, model, nprocs, fit_kwargs=None):
      if fit_kwargs is None:
          fit_kwargs = {}
  
      envs = schedule_processors(nprocs)
  
      from tensorflow.python.distribute.coordinator.values import (
          serialize_dataset_to_graph,
      )
  
      train_dataset = fit_kwargs.pop("x")
      val_dataset = fit_kwargs.pop("validation_data")
  
      train_ds_def = serialize_dataset_to_graph(train_dataset).numpy()
      train_elem_spec = train_dataset.element_spec
  
      if val_dataset is not None:
          val_ds_def = serialize_dataset_to_graph(val_dataset).numpy()
          val_elem_spec = val_dataset.element_spec
      else:
          val_ds_def = None
          val_elem_spec = None
  
      # this is to work around a tensorflow bug: https://github.com/keras-team/keras/issues/16023
      model.evaluate(train_dataset, verbose=0, steps=1)
      invalidInputError(
          model.compiled_metrics.built,
          "model.compiled_metrics.built should be True",
      )
      with TemporaryDirectory() as temp_dir:
          model.save(os.path.join(temp_dir, "temp_model"))
  
          train_args = (
              temp_dir,
              train_ds_def,
              train_elem_spec,
              val_ds_def,
              val_elem_spec,
              fit_kwargs,
          )
  
          histrories = backend.run(
              target=train_func,
              args=train_args,
              nprocs=nprocs,
              envs=envs,
          )
          model.load_weights(
              os.path.join(temp_dir, "trained_model_weights")
          )
      return histrories[0]
  
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 4 for calling a global function with example usage: distributed_train_keras(backend, model, nprocs, fit_kwargs) and returns: histrories