class_docstr: Local estimator based trainer executor used by the TFX Trainer component.
  
  How to create a trainer callback function to be used by this Trainer executor:
  An estimator can be executed by TFX by first creating a trainer_fn callback
  method that returns an estimator and some additional parameters, similar to
  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_utils.py#L285.
  This becomes the basis of the new Executor for Trainer. This Executor will
  then train and evaluate this estimator using the
  tf.estimator.train_and_evaluate API to train locally.
class_docstr_tok: ["Local","estimator","based","trainer","executor","used","by","the","TFX","Trainer","component",".","How","to","create","a","trainer","callback","function","to","be","used","by","this","Trainer","executor",":","An","estimator","can","be","executed","by","TFX","by","first","creating","a","trainer_fn","callback","method","that","returns","an","estimator","and","some","additional","parameters",",","similar","to","https",":","\/\/github.com\/tensorflow\/tfx\/blob\/master\/tfx\/examples\/chicago_taxi_pipeline\/taxi_utils.py","#","L285",".","This","becomes","the","basis","of","the","new","Executor","for","Trainer",".","This","Executor","will","then","train","and","evaluate","this","estimator","using","the","tf.estimator.train_and_evaluate","API","to","train","locally","."]
docstr: Uses a user-supplied tf.estimator to train a TensorFlow model locally.
  
  The Trainer Executor invokes a training_fn callback function provided by
  the user via the module_file parameter.  With the tf.estimator returned by
  this function, the Trainer Executor then builds a TensorFlow model using the
  user-provided tf.estimator.
  
  Args:
    input_dict: Input dict from input key to a list of ML-Metadata Artifacts.
      - examples: Examples used for training, must include 'train' and 'eval'
        if custom splits is not specified in train_args and eval_args.
      - transform_graph: Optional input transform graph.
      - schema: Schema of the data.
    output_dict: Output dict from output key to a list of Artifacts.
      - model: Exported model.
      - model_run: Model training related outputs (e.g., Tensorboard logs)
    exec_properties: A dict of execution properties.
      - train_args: JSON string of trainer_pb2.TrainArgs instance, providing
        args for training.
      - eval_args: JSON string of trainer_pb2.EvalArgs instance, providing
        args for eval.
      - module_file: Python module file containing UDF model definition.
        Exactly one of `module_file`, `module_path` and `trainer_fn` should
        be passed.
      - module_path: Python module path containing UDF model definition.
        Exactly one of `module_file`, `module_path` and `trainer_fn` should
        be passed.
      - trainer_fn: Python module path to the trainer function.
        Exactly one of `module_file`, `module_path` and `trainer_fn` should
        be passed.
      - warm_starting: Whether or not we need to do warm starting.
      - warm_start_from: Optional. If warm_starting is True, this is the
        directory to find previous model to warm start on.
      - custom_config: Optional. JSON-serialized dict of additional parameters
        to pass to trainer function.
  
  Returns:
    None
  
  Raises:
    ValueError: When not exactly one of `module_file`, `module_path` and
      `trainer_fn` are present in `exec_properties`.
docstr_tok: ["Uses","a","user-supplied","tf.estimator","to","train","a","TensorFlow","model","locally",".","The","Trainer","Executor","invokes","a","training_fn","callback","function","provided","by","the","user","via","the","module_file","parameter",".","With","the","tf.estimator","returned","by","this","function",",","the","Trainer","Executor","then","builds","a","TensorFlow","model","using","the","user-provided","tf.estimator",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","ML-Metadata","Artifacts",".","-","examples",":","Examples","used","for","training",",","must","include","'train","'","and","'eval'","if","custom","splits","is","not","specified","in","train_args","and","eval_args",".","-","transform_graph",":","Optional","input","transform","graph",".","-","schema",":","Schema","of","the","data",".","output_dict",":","Output","dict","from","output","key","to","a","list","of","Artifacts",".","-","model",":","Exported","model",".","-","model_run",":","Model","training","related","outputs","(","e.g.",",","Tensorboard","logs",")","exec_properties",":","A","dict","of","execution","properties",".","-","train_args",":","JSON","string","of","trainer_pb2.TrainArgs","instance",",","providing","args","for","training",".","-","eval_args",":","JSON","string","of","trainer_pb2.EvalArgs","instance",",","providing","args","for","eval",".","-","module_file",":","Python","module","file","containing","UDF","model","definition",".","Exactly","one","of","`","module_file","`",",","`","module_path","`","and","`","trainer_fn","`","should","be","passed",".","-","module_path",":","Python","module","path","containing","UDF","model","definition",".","Exactly","one","of","`","module_file","`",",","`","module_path","`","and","`","trainer_fn","`","should","be","passed",".","-","trainer_fn",":","Python","module","path","to","the","trainer","function",".","Exactly","one","of","`","module_file","`",",","`","module_path","`","and","`","trainer_fn","`","should","be","passed",".","-","warm_starting",":","Whether","or","not","we","need","to","do","warm","starting",".","-","warm_start_from",":","Optional",".","If","warm_starting","is","True",",","this","is","the","directory","to","find","previous","model","to","warm","start","on",".","-","custom_config",":","Optional",".","JSON-serialized","dict","of","additional","parameters","to","pass","to","trainer","function",".","Returns",":","None","Raises",":","ValueError",":","When","not","exactly","one","of","`","module_file","`",",","`","module_path","`","and","`","trainer_fn","`","are","present","in","`","exec_properties","`","."]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Uses a user-supplied tf.estimator to train a TensorFlow model locally.
  
      The Trainer Executor invokes a training_fn callback function provided by
      the user via the module_file parameter.  With the tf.estimator returned by
      this function, the Trainer Executor then builds a TensorFlow model using the
      user-provided tf.estimator.
  
      Args:
        input_dict: Input dict from input key to a list of ML-Metadata Artifacts.
          - examples: Examples used for training, must include 'train' and 'eval'
            if custom splits is not specified in train_args and eval_args.
          - transform_graph: Optional input transform graph.
          - schema: Schema of the data.
        output_dict: Output dict from output key to a list of Artifacts.
          - model: Exported model.
          - model_run: Model training related outputs (e.g., Tensorboard logs)
        exec_properties: A dict of execution properties.
          - train_args: JSON string of trainer_pb2.TrainArgs instance, providing
            args for training.
          - eval_args: JSON string of trainer_pb2.EvalArgs instance, providing
            args for eval.
          - module_file: Python module file containing UDF model definition.
            Exactly one of `module_file`, `module_path` and `trainer_fn` should
            be passed.
          - module_path: Python module path containing UDF model definition.
            Exactly one of `module_file`, `module_path` and `trainer_fn` should
            be passed.
          - trainer_fn: Python module path to the trainer function.
            Exactly one of `module_file`, `module_path` and `trainer_fn` should
            be passed.
          - warm_starting: Whether or not we need to do warm starting.
          - warm_start_from: Optional. If warm_starting is True, this is the
            directory to find previous model to warm start on.
          - custom_config: Optional. JSON-serialized dict of additional parameters
            to pass to trainer function.
  
      Returns:
        None
  
      Raises:
        ValueError: When not exactly one of `module_file`, `module_path` and
          `trainer_fn` are present in `exec_properties`.
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      fn_args = self._GetFnArgs(
          input_dict, output_dict, exec_properties
      )
      trainer_fn = udf_utils.get_fn(exec_properties, "trainer_fn")
  
      schema = io_utils.parse_pbtxt_file(
          fn_args.schema_file, schema_pb2.Schema()
      )
  
      # TODO(b/160795287): Deprecate estimator based executor.
      # Provide user with a modified fn_args, with model_run given as
      # the working directory. Executor will then copy user models to
      # model artifact directory.
      serving_dest = fn_args.serving_model_dir
      eval_dest = fn_args.eval_model_dir
  
      working_dir = fn_args.model_run_dir
      fn_args.serving_model_dir = path_utils.serving_model_dir(
          working_dir
      )
      fn_args.eval_model_dir = path_utils.eval_model_dir(working_dir)
  
      training_spec = trainer_fn(fn_args, schema)
  
      # Train the model
      absl.logging.info("Training model.")
      tf_estimator.train_and_evaluate(
          training_spec["estimator"],
          training_spec["train_spec"],
          training_spec["eval_spec"],
      )
  
      absl.logging.info(
          "Training complete. Model written to %s. ModelRun written to %s",
          fn_args.serving_model_dir,
          fn_args.model_run_dir,
      )
  
      # Export an eval savedmodel for TFMA. If distributed training, it must only
      # be written by the chief worker, as would be done for serving savedmodel.
      if _is_chief():
          absl.logging.info("Exporting eval_savedmodel for TFMA.")
          tfma.export.export_eval_savedmodel(
              estimator=training_spec["estimator"],
              export_dir_base=fn_args.eval_model_dir,
              eval_input_receiver_fn=training_spec[
                  "eval_input_receiver_fn"
              ],
          )
  
          absl.logging.info(
              "Exported eval_savedmodel to %s.", fn_args.eval_model_dir
          )
  
          # TODO(b/160795287): Deprecate estimator based executor.
          # Copy serving and eval model from model_run to model artifact directory.
          serving_source = path_utils.serving_model_path(
              fn_args.model_run_dir
          )
          io_utils.copy_dir(serving_source, serving_dest)
          absl.logging.info(
              "Serving model copied to: %s.", serving_dest
          )
  
          eval_source = path_utils.eval_model_path(
              fn_args.model_run_dir
          )
          io_utils.copy_dir(eval_source, eval_dest)
          absl.logging.info("Eval model copied to: %s.", eval_dest)
  
      else:
          absl.logging.info(
              "Model export is skipped because this is not the chief worker."
          )
  
imports: import json
  import os
  from typing import Any, Dict, List
  import absl
  from tensorflow import estimator
  import tensorflow_model_analysis
  from tfx import types
  from tfx.components.trainer import constants
  from tfx.components.trainer import fn_args_utils
  from tfx.components.util import udf_utils
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.io import fileio
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import deprecation_utils
  from tfx.utils import io_utils
  from tfx.utils import path_utils
  from tensorflow.python.lib.io import file_io
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 1 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types