code: def run_ncf(FLAGS):
      """Run NCF training and eval with Keras."""
  
      #########################################################################
      # Construct AutoDist with ResourceSpec for Different Strategies
      resource_spec_file = os.path.join(
          os.path.dirname(__file__), "../resource_spec.yml"
      )
      resource_info = yaml.safe_load(open(resource_spec_file, "r"))
      try:
          node_num = len(resource_info["nodes"])
      except ValueError:
          print("nodes need to be set in specficiation file")
  
      try:
          gpu_num = len(resource_info["nodes"][0]["gpus"])
      except ValueError:
          print("gpus need to be set in specficiation file")
  
      if FLAGS.autodist_patch_tf:
          os.environ["AUTODIST_PATCH_TF"] = "1"
      else:
          os.environ["AUTODIST_PATCH_TF"] = "0"
  
      if FLAGS.proxy:
          local_proxy_variable = True
      else:
          local_proxy_variable = False
  
      if FLAGS.autodist_strategy == "PS":
          autodist = AutoDist(
              resource_spec_file,
              PS(local_proxy_variable=local_proxy_variable),
          )
      elif FLAGS.autodist_strategy == "PSLoadBalancing":
          autodist = AutoDist(
              resource_spec_file,
              PSLoadBalancing(
                  local_proxy_variable=local_proxy_variable
              ),
          )
      elif FLAGS.autodist_strategy == "PartitionedPS":
          autodist = AutoDist(
              resource_spec_file,
              PartitionedPS(local_proxy_variable=local_proxy_variable),
          )
      elif FLAGS.autodist_strategy == "AllReduce":
          autodist = AutoDist(
              resource_spec_file, AllReduce(chunk_size=256)
          )
      elif FLAGS.autodist_strategy == "Parallax":
          autodist = AutoDist(
              resource_spec_file,
              Parallax(
                  chunk_size=256,
                  local_proxy_variable=local_proxy_variable,
              ),
          )
      else:
          raise ValueError(
              "the strategy can be only from PS, PSLoadBalancing, PartitionedPS, AllReduce, Parallax"
          )
      #########################################################################
      if FLAGS.seed is not None:
          print("Setting tf seed")
          tf.random.set_seed(FLAGS.seed)
  
      model_helpers.apply_clean(FLAGS)
  
      if FLAGS.dtype == "fp16" and FLAGS.fp16_implementation == "keras":
          policy = tf.keras.mixed_precision.experimental.Policy(
              "mixed_float16",
              loss_scale=flags_core.get_loss_scale(
                  FLAGS, default_for_fp16="dynamic"
              ),
          )
          tf.keras.mixed_precision.experimental.set_policy(policy)
  
      params = ncf_common.parse_flags(FLAGS)
      params["distribute_strategy"] = None
  
      batch_size = params["batch_size"]
      time_callback = keras_utils.TimeHistory(
          batch_size, FLAGS.log_steps
      )
      callbacks = [time_callback]
  
      producer, input_meta_data = None, None
      generate_input_online = params["train_dataset_path"] is None
  
      if generate_input_online:
          num_users, num_items, _, _, producer = ncf_common.get_inputs(
              params
          )
          producer.start()
          per_epoch_callback = IncrementEpochCallback(producer)
          callbacks.append(per_epoch_callback)
      else:
          assert (
              params["eval_dataset_path"]
              and params["input_meta_data_path"]
          )
          with tf.io.gfile.GFile(
              params["input_meta_data_path"], "rb"
          ) as reader:
              input_meta_data = json.loads(
                  reader.read().decode("utf-8")
              )
              num_users = input_meta_data["num_users"]
              num_items = input_meta_data["num_items"]
  
      params["num_users"], params["num_items"] = num_users, num_items
  
      if FLAGS.early_stopping:
          early_stopping_callback = CustomEarlyStopping(
              "val_HR_METRIC", desired_value=FLAGS.hr_threshold
          )
          callbacks.append(early_stopping_callback)
  
      with tf.Graph().as_default(), autodist.scope():
          (
              train_input_dataset,
              eval_input_dataset,
              num_train_steps,
              num_eval_steps,
          ) = ncf_input_pipeline.create_ncf_input_data(
              params, producer, input_meta_data, None
          )
          steps_per_epoch = (
              None if generate_input_online else num_train_steps
          )
          keras_model = _get_keras_model(params)
          if FLAGS.optimizer == "adam":
              optimizer = tf.keras.optimizers.Adam(
                  learning_rate=params["learning_rate"],
                  beta_1=params["beta1"],
                  beta_2=params["beta2"],
                  epsilon=params["epsilon"],
              )
          elif FLAGS.optimizer == "sgd":
              optimizer = tf.keras.optimizers.SGD(
                  learning_rate=params["learning_rate"]
              )
          elif FLAGS.optimizer == "lazyadam":
              optimizer = LazyAdam(
                  learning_rate=params["learning_rate"],
                  beta_1=params["beta1"],
                  beta_2=params["beta2"],
                  epsilon=params["epsilon"],
              )
          else:
              raise ValueError("Do not support other optimizers...")
          if FLAGS.fp16_implementation == "graph_rewrite":
              optimizer = tf.compat.v1.train.experimental.enable_mixed_precision_graph_rewrite(
                  optimizer,
                  loss_scale=flags_core.get_loss_scale(
                      FLAGS, default_for_fp16="dynamic"
                  ),
              )
          elif FLAGS.dtype == "fp16" and params["keras_use_ctl"]:
              optimizer = tf.keras.mixed_precision.experimental.LossScaleOptimizer(
                  optimizer,
                  tf.keras.mixed_precision.experimental.global_policy().loss_scale,
              )
  
          return run_ncf_custom_training(
              params,
              autodist,
              keras_model,
              optimizer,
              callbacks,
              train_input_dataset,
              eval_input_dataset,
              num_train_steps,
              num_eval_steps,
              generate_input_online=generate_input_online,
              return_simulation=FLAGS.simulation_strategy_id
              is not None,
          )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import json
  import yaml
  import os
  import sys
  import tensorflow
  from absl import app
  from absl import flags
  from absl import logging
  from os.path import expanduser
  from tensorflow_addons.optimizers import LazyAdam
  from utils.recommendation import constants
  from utils.recommendation import movielens
  from utils.recommendation import ncf_common
  from utils.recommendation import ncf_input_pipeline
  from utils.recommendation import neumf_model
  from utils.flags import core
  from utils.logs import logger
  from utils.logs import mlperf_helper
  from utils.misc import keras_utils
  from utils.misc import model_helpers
  from autodist import AutoDist
  from autodist.strategy.ps_strategy import PS
  from autodist.strategy.ps_lb_strategy import PSLoadBalancing
  from autodist.strategy.partitioned_ps_strategy import PartitionedPS
  from autodist.strategy.all_reduce_strategy import AllReduce
  from autodist.strategy.parallax_strategy import Parallax
label_desc: Use image node_id 2 for calling a global function with example usage: run_ncf(FLAGS) and returns: run_ncf_custom_training