mapPartitions function to run single-node inferencing from a checkpoint/saved_model, using the model's input/output mappings.

  Args:
    :iterator: input RDD partition iterator.
    :args: arguments for TFModel, in argparse format
    :tf_args: arguments for TensorFlow inferencing code, in argparse or ARGV format.

  Returns:
    An iterator of result data.

def _run_model(iterator, args, tf_args):
  """mapPartitions function to run single-node inferencing from a checkpoint/saved_model, using the model's input/output mappings.

  Args:
    :iterator: input RDD partition iterator.
    :args: arguments for TFModel, in argparse format
    :tf_args: arguments for TensorFlow inferencing code, in argparse or ARGV format.

  Returns:
    An iterator of result data.
  """
  single_node_env(tf_args)

  logging.info("===== input_mapping: {}".format(args.input_mapping))
  logging.info("===== output_mapping: {}".format(args.output_mapping))
  input_tensor_names = [tensor for col, tensor in sorted(args.input_mapping.items())]
  output_tensor_names = [tensor for tensor, col in sorted(args.output_mapping.items())]

  # if using a signature_def_key, get input/output tensor info from the requested signature
  if args.signature_def_key:
    assert args.export_dir, "Inferencing with signature_def_key requires --export_dir argument"
    logging.info("===== loading meta_graph_def for tag_set ({0}) from saved_model: {1}".format(args.tag_set, args.export_dir))
    meta_graph_def = get_meta_graph_def(args.export_dir, args.tag_set)
    signature = meta_graph_def.signature_def[args.signature_def_key]
    logging.debug("signature: {}".format(signature))
    inputs_tensor_info = signature.inputs
    logging.debug("inputs_tensor_info: {0}".format(inputs_tensor_info))
    outputs_tensor_info = signature.outputs
    logging.debug("outputs_tensor_info: {0}".format(outputs_tensor_info))

  result = []

  global global_sess, global_args
  if global_sess and global_args == args:
    # if graph/session already loaded/started (and using same args), just reuse it
    sess = global_sess
  else:
    # otherwise, create new session and load graph from disk
    tf.reset_default_graph()
    sess = tf.Session(graph=tf.get_default_graph())
    if args.export_dir:
      assert args.tag_set, "Inferencing from a saved_model requires --tag_set"
      # load graph from a saved_model
      logging.info("===== restoring from saved_model: {}".format(args.export_dir))
      loader.load(sess, args.tag_set.split(','), args.export_dir)
    elif args.model_dir:
      # load graph from a checkpoint
      ckpt = tf.train.latest_checkpoint(args.model_dir)
      assert ckpt, "Invalid model checkpoint path: {}".format(args.model_dir)
      logging.info("===== restoring from checkpoint: {}".format(ckpt + ".meta"))
      saver = tf.train.import_meta_graph(ckpt + ".meta", clear_devices=True)
      saver.restore(sess, ckpt)
    else:
      raise Exception("Inferencing requires either --model_dir or --export_dir argument")
    global_sess = sess
    global_args = args

  # get list of input/output tensors (by name)
  if args.signature_def_key:
    input_tensors = [inputs_tensor_info[t].name for t in input_tensor_names]
    output_tensors = [outputs_tensor_info[output_tensor_names[0]].name]
  else:
    input_tensors = [t + ':0' for t in input_tensor_names]
    output_tensors = [t + ':0' for t in output_tensor_names]

  logging.info("input_tensors: {0}".format(input_tensors))
  logging.info("output_tensors: {0}".format(output_tensors))

  # feed data in batches and return output tensors
  for tensors in yield_batch(iterator, args.batch_size, len(input_tensor_names)):
    inputs_feed_dict = {}
    for i in range(len(input_tensors)):
      inputs_feed_dict[input_tensors[i]] = tensors[i]

    outputs = sess.run(output_tensors, feed_dict=inputs_feed_dict)
    lengths = [len(output) for output in outputs]
    input_size = len(tensors[0])
    assert all([length == input_size for length in lengths]), "Output array sizes {} must match input size: {}".format(lengths, input_size)
    python_outputs = [output.tolist() for output in outputs]      # convert from numpy to standard python types
    result.extend(zip(*python_outputs))                           # convert to an array of tuples of "output columns"

  return result