We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
`transform_output` not truely optional in 0.14.0 release
`transform_output` not truely optional for Trainer in 0.14.0 release. Culprit code is [here](https://github.com/tensorflow/tfx/blob/r0.14/tfx/components/trainer/executor.py#L148).

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tfx/components/trainer/executor.py`
Content:
```
1 # Copyright 2019 Google LLC. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """TFX local trainer executor."""
15 
16 from __future__ import absolute_import
17 from __future__ import division
18 from __future__ import print_function
19 
20 import os
21 import tensorflow as tf
22 import tensorflow_model_analysis as tfma
23 from typing import Any, Dict, List, Text
24 
25 from tensorflow_metadata.proto.v0 import schema_pb2
26 from tfx import types
27 from tfx.components.base import base_executor
28 from tfx.extensions.google_cloud_ai_platform import runner
29 from tfx.proto import trainer_pb2
30 from tfx.types import artifact_utils
31 from tfx.utils import import_utils
32 from tfx.utils import io_utils
33 from tfx.utils import path_utils
34 from google.protobuf import json_format
35 
36 
37 def _all_files_pattern(file_pattern: Text) -> Text:
38   return '{}*'.format(file_pattern)
39 
40 
41 class Executor(base_executor.BaseExecutor):
42   """Local trainer used by the TFX Trainer component.
43 
44   The Trainer executor supplements TensorFlow training with a component to
45   enable warm-start training of any user-specified tf.estimator. The Trainer is
46   a library built on top of TensorFlow that is expected to be integrated into a
47   custom user-specified binary.
48 
49   To include Trainer in a TFX pipeline, configure your pipeline similar to
50   https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple.py#L104.
51 
52   For more details on the Trainer component itself, please refer to
53   https://tensorflow.org/tfx/guide/trainer.  For a tutorial on TF Estimator,
54   please refer to https://www.tensorflow.org/extend/estimators.
55 
56   How to create a trainer callback function to be used by this Trainer executor:
57   An estimator can be executed by TFX by first creating a trainer_fn callback
58   method that returns an estimator and some additional parameters, similar to
59   https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_utils.py#L285.
60   This becomes the basis of the new Executor for Trainer. This Executor will
61   then train and evaluate this estimator using the
62   tf.estimator.train_and_evaluate API to train locally.
63   """
64 
65   # Name of subdirectory which contains checkpoints from prior runs
66   _CHECKPOINT_FILE_NAME = 'checkpoint'
67 
68   def _GetTrainerFn(self, exec_properties: Dict[Text, Any]) -> Any:
69     """Loads and returns user-defined trainer_fn."""
70 
71     has_module_file = bool(exec_properties.get('module_file'))
72     has_trainer_fn = bool(exec_properties.get('trainer_fn'))
73 
74     if has_module_file == has_trainer_fn:
75       raise ValueError(
76           "Neither or both of 'module_file' 'trainer_fn' have been supplied in "
77           "'exec_properties'.")
78 
79     if has_module_file:
80       return import_utils.import_func_from_source(
81           exec_properties['module_file'], 'trainer_fn')
82 
83     trainer_fn_path_split = exec_properties['trainer_fn'].split('.')
84     return import_utils.import_func_from_module(
85         '.'.join(trainer_fn_path_split[0:-1]), trainer_fn_path_split[-1])
86 
87   def Do(self, input_dict: Dict[Text, List[types.Artifact]],
88          output_dict: Dict[Text, List[types.Artifact]],
89          exec_properties: Dict[Text, Any]) -> None:
90     """Uses a user-supplied tf.estimator to train a TensorFlow model locally.
91 
92     The Trainer Executor invokes a training_fn callback function provided by
93     the user via the module_file parameter.  With the tf.estimator returned by
94     this function, the Trainer Executor then builds a TensorFlow model using the
95     user-provided tf.estimator.
96 
97     Args:
98       input_dict: Input dict from input key to a list of ML-Metadata Artifacts.
99         - examples: Examples used for training, must include 'train' and 'eval'
100           splits.
101         - transform_output: Optional input transform graph.
102         - schema: Schema of the data.
103       output_dict: Output dict from output key to a list of Artifacts.
104         - output: Exported model.
105       exec_properties: A dict of execution properties.
106         - train_args: JSON string of trainer_pb2.TrainArgs instance, providing
107           args for training.
108         - eval_args: JSON string of trainer_pb2.EvalArgs instance, providing
109           args for eval.
110         - module_file: Python module file containing UDF model definition.
111         - warm_starting: Whether or not we need to do warm starting.
112         - warm_start_from: Optional. If warm_starting is True, this is the
113           directory to find previous model to warm start on.
114 
115     Returns:
116       None
117 
118     Raises:
119       ValueError: When neither or both of 'module_file' and 'trainer_fn'
120         are present in 'exec_properties'.
121     """
122     self._log_startup(input_dict, output_dict, exec_properties)
123 
124     # TODO(zhitaoli): Deprecate this in a future version.
125     if exec_properties.get('custom_config', None):
126       cmle_args = exec_properties.get('custom_config',
127                                       {}).get('cmle_training_args')
128       if cmle_args:
129         executor_class_path = '.'.join([Executor.__module__, Executor.__name__])
130         tf.logging.warn(
131             'Passing \'cmle_training_args\' to trainer directly is deprecated, '
132             'please use extension executor at '
133             'tfx.extensions.google_cloud_ai_platform.trainer.executor instead')
134 
135         return runner.start_cmle_training(input_dict, output_dict,
136                                           exec_properties, executor_class_path,
137                                           cmle_args)
138 
139     trainer_fn = self._GetTrainerFn(exec_properties)
140 
141     # Set up training parameters
142     train_files = [
143         _all_files_pattern(
144             artifact_utils.get_split_uri(input_dict['examples'], 'train'))
145     ]
146     transform_output = artifact_utils.get_single_uri(
147         input_dict['transform_output']
148     ) if input_dict['transform_output'] else None
149     eval_files = [
150         _all_files_pattern(
151             artifact_utils.get_split_uri(input_dict['examples'], 'eval'))
152     ]
153     schema_file = io_utils.get_only_uri_in_dir(
154         artifact_utils.get_single_uri(input_dict['schema']))
155 
156     train_args = trainer_pb2.TrainArgs()
157     eval_args = trainer_pb2.EvalArgs()
158     json_format.Parse(exec_properties['train_args'], train_args)
159     json_format.Parse(exec_properties['eval_args'], eval_args)
160 
161     # https://github.com/tensorflow/tfx/issues/45: Replace num_steps=0 with
162     # num_steps=None.  Conversion of the proto to python will set the default
163     # value of an int as 0 so modify the value here.  Tensorflow will raise an
164     # error if num_steps <= 0.
165     train_steps = train_args.num_steps or None
166     eval_steps = eval_args.num_steps or None
167 
168     output_path = artifact_utils.get_single_uri(output_dict['output'])
169     serving_model_dir = path_utils.serving_model_dir(output_path)
170     eval_model_dir = path_utils.eval_model_dir(output_path)
171 
172     # Assemble warm start path if needed.
173     warm_start_from = None
174     if exec_properties.get('warm_starting') and exec_properties.get(
175         'warm_start_from'):
176       previous_model_dir = os.path.join(exec_properties['warm_start_from'],
177                                         path_utils.SERVING_MODEL_DIR)
178       if previous_model_dir and tf.gfile.Exists(
179           os.path.join(previous_model_dir, self._CHECKPOINT_FILE_NAME)):
180         warm_start_from = previous_model_dir
181 
182     # TODO(b/126242806) Use PipelineInputs when it is available in third_party.
183     hparams = tf.contrib.training.HParams(
184         # A list of uris for train files.
185         train_files=train_files,
186         # An optional single uri for transform graph produced by TFT. Will be
187         # None if not specified.
188         transform_output=transform_output,
189         # A single uri for the output directory of the serving model.
190         serving_model_dir=serving_model_dir,
191         # A list of uris for eval files.
192         eval_files=eval_files,
193         # A single uri for schema file.
194         schema_file=schema_file,
195         # Number of train steps.
196         train_steps=train_steps,
197         # Number of eval steps.
198         eval_steps=eval_steps,
199         # A single uri for the model directory to warm start from.
200         warm_start_from=warm_start_from)
201 
202     schema = io_utils.parse_pbtxt_file(schema_file, schema_pb2.Schema())
203 
204     training_spec = trainer_fn(hparams, schema)
205 
206     # Train the model
207     tf.logging.info('Training model.')
208     tf.estimator.train_and_evaluate(training_spec['estimator'],
209                                     training_spec['train_spec'],
210                                     training_spec['eval_spec'])
211     tf.logging.info('Training complete.  Model written to %s',
212                     serving_model_dir)
213 
214     # Export an eval savedmodel for TFMA
215     tf.logging.info('Exporting eval_savedmodel for TFMA.')
216     tfma.export.export_eval_savedmodel(
217         estimator=training_spec['estimator'],
218         export_dir_base=eval_model_dir,
219         eval_input_receiver_fn=training_spec['eval_input_receiver_fn'])
220 
221     tf.logging.info('Exported eval_savedmodel to %s.', eval_model_dir)
222
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/tfx/components/trainer/executor.py b/tfx/components/trainer/executor.py
--- a/tfx/components/trainer/executor.py
+++ b/tfx/components/trainer/executor.py
@@ -144,8 +144,8 @@
             artifact_utils.get_split_uri(input_dict['examples'], 'train'))
     ]
     transform_output = artifact_utils.get_single_uri(
-        input_dict['transform_output']
-    ) if input_dict['transform_output'] else None
+        input_dict['transform_output']) if input_dict.get(
+            'transform_output', None) else None
     eval_files = [
         _all_files_pattern(
             artifact_utils.get_split_uri(input_dict['examples'], 'eval'))

{"golden_diff": "diff --git a/tfx/components/trainer/executor.py b/tfx/components/trainer/executor.py\n--- a/tfx/components/trainer/executor.py\n+++ b/tfx/components/trainer/executor.py\n@@ -144,8 +144,8 @@\n             artifact_utils.get_split_uri(input_dict['examples'], 'train'))\n     ]\n     transform_output = artifact_utils.get_single_uri(\n-        input_dict['transform_output']\n-    ) if input_dict['transform_output'] else None\n+        input_dict['transform_output']) if input_dict.get(\n+            'transform_output', None) else None\n     eval_files = [\n         _all_files_pattern(\n             artifact_utils.get_split_uri(input_dict['examples'], 'eval'))\n", "issue": "`transform_output` not truely optional in 0.14.0 release\n`transform_output` not truely optional for Trainer in 0.14.0 release. Culprit code is [here](https://github.com/tensorflow/tfx/blob/r0.14/tfx/components/trainer/executor.py#L148).\n", "before_files": [{"content": "# Copyright 2019 Google LLC. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"TFX local trainer executor.\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport os\nimport tensorflow as tf\nimport tensorflow_model_analysis as tfma\nfrom typing import Any, Dict, List, Text\n\nfrom tensorflow_metadata.proto.v0 import schema_pb2\nfrom tfx import types\nfrom tfx.components.base import base_executor\nfrom tfx.extensions.google_cloud_ai_platform import runner\nfrom tfx.proto import trainer_pb2\nfrom tfx.types import artifact_utils\nfrom tfx.utils import import_utils\nfrom tfx.utils import io_utils\nfrom tfx.utils import path_utils\nfrom google.protobuf import json_format\n\n\ndef _all_files_pattern(file_pattern: Text) -> Text:\n  return '{}*'.format(file_pattern)\n\n\nclass Executor(base_executor.BaseExecutor):\n  \"\"\"Local trainer used by the TFX Trainer component.\n\n  The Trainer executor supplements TensorFlow training with a component to\n  enable warm-start training of any user-specified tf.estimator. The Trainer is\n  a library built on top of TensorFlow that is expected to be integrated into a\n  custom user-specified binary.\n\n  To include Trainer in a TFX pipeline, configure your pipeline similar to\n  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple.py#L104.\n\n  For more details on the Trainer component itself, please refer to\n  https://tensorflow.org/tfx/guide/trainer.  For a tutorial on TF Estimator,\n  please refer to https://www.tensorflow.org/extend/estimators.\n\n  How to create a trainer callback function to be used by this Trainer executor:\n  An estimator can be executed by TFX by first creating a trainer_fn callback\n  method that returns an estimator and some additional parameters, similar to\n  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_utils.py#L285.\n  This becomes the basis of the new Executor for Trainer. This Executor will\n  then train and evaluate this estimator using the\n  tf.estimator.train_and_evaluate API to train locally.\n  \"\"\"\n\n  # Name of subdirectory which contains checkpoints from prior runs\n  _CHECKPOINT_FILE_NAME = 'checkpoint'\n\n  def _GetTrainerFn(self, exec_properties: Dict[Text, Any]) -> Any:\n    \"\"\"Loads and returns user-defined trainer_fn.\"\"\"\n\n    has_module_file = bool(exec_properties.get('module_file'))\n    has_trainer_fn = bool(exec_properties.get('trainer_fn'))\n\n    if has_module_file == has_trainer_fn:\n      raise ValueError(\n          \"Neither or both of 'module_file' 'trainer_fn' have been supplied in \"\n          \"'exec_properties'.\")\n\n    if has_module_file:\n      return import_utils.import_func_from_source(\n          exec_properties['module_file'], 'trainer_fn')\n\n    trainer_fn_path_split = exec_properties['trainer_fn'].split('.')\n    return import_utils.import_func_from_module(\n        '.'.join(trainer_fn_path_split[0:-1]), trainer_fn_path_split[-1])\n\n  def Do(self, input_dict: Dict[Text, List[types.Artifact]],\n         output_dict: Dict[Text, List[types.Artifact]],\n         exec_properties: Dict[Text, Any]) -> None:\n    \"\"\"Uses a user-supplied tf.estimator to train a TensorFlow model locally.\n\n    The Trainer Executor invokes a training_fn callback function provided by\n    the user via the module_file parameter.  With the tf.estimator returned by\n    this function, the Trainer Executor then builds a TensorFlow model using the\n    user-provided tf.estimator.\n\n    Args:\n      input_dict: Input dict from input key to a list of ML-Metadata Artifacts.\n        - examples: Examples used for training, must include 'train' and 'eval'\n          splits.\n        - transform_output: Optional input transform graph.\n        - schema: Schema of the data.\n      output_dict: Output dict from output key to a list of Artifacts.\n        - output: Exported model.\n      exec_properties: A dict of execution properties.\n        - train_args: JSON string of trainer_pb2.TrainArgs instance, providing\n          args for training.\n        - eval_args: JSON string of trainer_pb2.EvalArgs instance, providing\n          args for eval.\n        - module_file: Python module file containing UDF model definition.\n        - warm_starting: Whether or not we need to do warm starting.\n        - warm_start_from: Optional. If warm_starting is True, this is the\n          directory to find previous model to warm start on.\n\n    Returns:\n      None\n\n    Raises:\n      ValueError: When neither or both of 'module_file' and 'trainer_fn'\n        are present in 'exec_properties'.\n    \"\"\"\n    self._log_startup(input_dict, output_dict, exec_properties)\n\n    # TODO(zhitaoli): Deprecate this in a future version.\n    if exec_properties.get('custom_config', None):\n      cmle_args = exec_properties.get('custom_config',\n                                      {}).get('cmle_training_args')\n      if cmle_args:\n        executor_class_path = '.'.join([Executor.__module__, Executor.__name__])\n        tf.logging.warn(\n            'Passing \\'cmle_training_args\\' to trainer directly is deprecated, '\n            'please use extension executor at '\n            'tfx.extensions.google_cloud_ai_platform.trainer.executor instead')\n\n        return runner.start_cmle_training(input_dict, output_dict,\n                                          exec_properties, executor_class_path,\n                                          cmle_args)\n\n    trainer_fn = self._GetTrainerFn(exec_properties)\n\n    # Set up training parameters\n    train_files = [\n        _all_files_pattern(\n            artifact_utils.get_split_uri(input_dict['examples'], 'train'))\n    ]\n    transform_output = artifact_utils.get_single_uri(\n        input_dict['transform_output']\n    ) if input_dict['transform_output'] else None\n    eval_files = [\n        _all_files_pattern(\n            artifact_utils.get_split_uri(input_dict['examples'], 'eval'))\n    ]\n    schema_file = io_utils.get_only_uri_in_dir(\n        artifact_utils.get_single_uri(input_dict['schema']))\n\n    train_args = trainer_pb2.TrainArgs()\n    eval_args = trainer_pb2.EvalArgs()\n    json_format.Parse(exec_properties['train_args'], train_args)\n    json_format.Parse(exec_properties['eval_args'], eval_args)\n\n    # https://github.com/tensorflow/tfx/issues/45: Replace num_steps=0 with\n    # num_steps=None.  Conversion of the proto to python will set the default\n    # value of an int as 0 so modify the value here.  Tensorflow will raise an\n    # error if num_steps <= 0.\n    train_steps = train_args.num_steps or None\n    eval_steps = eval_args.num_steps or None\n\n    output_path = artifact_utils.get_single_uri(output_dict['output'])\n    serving_model_dir = path_utils.serving_model_dir(output_path)\n    eval_model_dir = path_utils.eval_model_dir(output_path)\n\n    # Assemble warm start path if needed.\n    warm_start_from = None\n    if exec_properties.get('warm_starting') and exec_properties.get(\n        'warm_start_from'):\n      previous_model_dir = os.path.join(exec_properties['warm_start_from'],\n                                        path_utils.SERVING_MODEL_DIR)\n      if previous_model_dir and tf.gfile.Exists(\n          os.path.join(previous_model_dir, self._CHECKPOINT_FILE_NAME)):\n        warm_start_from = previous_model_dir\n\n    # TODO(b/126242806) Use PipelineInputs when it is available in third_party.\n    hparams = tf.contrib.training.HParams(\n        # A list of uris for train files.\n        train_files=train_files,\n        # An optional single uri for transform graph produced by TFT. Will be\n        # None if not specified.\n        transform_output=transform_output,\n        # A single uri for the output directory of the serving model.\n        serving_model_dir=serving_model_dir,\n        # A list of uris for eval files.\n        eval_files=eval_files,\n        # A single uri for schema file.\n        schema_file=schema_file,\n        # Number of train steps.\n        train_steps=train_steps,\n        # Number of eval steps.\n        eval_steps=eval_steps,\n        # A single uri for the model directory to warm start from.\n        warm_start_from=warm_start_from)\n\n    schema = io_utils.parse_pbtxt_file(schema_file, schema_pb2.Schema())\n\n    training_spec = trainer_fn(hparams, schema)\n\n    # Train the model\n    tf.logging.info('Training model.')\n    tf.estimator.train_and_evaluate(training_spec['estimator'],\n                                    training_spec['train_spec'],\n                                    training_spec['eval_spec'])\n    tf.logging.info('Training complete.  Model written to %s',\n                    serving_model_dir)\n\n    # Export an eval savedmodel for TFMA\n    tf.logging.info('Exporting eval_savedmodel for TFMA.')\n    tfma.export.export_eval_savedmodel(\n        estimator=training_spec['estimator'],\n        export_dir_base=eval_model_dir,\n        eval_input_receiver_fn=training_spec['eval_input_receiver_fn'])\n\n    tf.logging.info('Exported eval_savedmodel to %s.', eval_model_dir)\n", "path": "tfx/components/trainer/executor.py"}], "after_files": [{"content": "# Copyright 2019 Google LLC. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"TFX local trainer executor.\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport os\nimport tensorflow as tf\nimport tensorflow_model_analysis as tfma\nfrom typing import Any, Dict, List, Text\n\nfrom tensorflow_metadata.proto.v0 import schema_pb2\nfrom tfx import types\nfrom tfx.components.base import base_executor\nfrom tfx.extensions.google_cloud_ai_platform import runner\nfrom tfx.proto import trainer_pb2\nfrom tfx.types import artifact_utils\nfrom tfx.utils import import_utils\nfrom tfx.utils import io_utils\nfrom tfx.utils import path_utils\nfrom google.protobuf import json_format\n\n\ndef _all_files_pattern(file_pattern: Text) -> Text:\n  return '{}*'.format(file_pattern)\n\n\nclass Executor(base_executor.BaseExecutor):\n  \"\"\"Local trainer used by the TFX Trainer component.\n\n  The Trainer executor supplements TensorFlow training with a component to\n  enable warm-start training of any user-specified tf.estimator. The Trainer is\n  a library built on top of TensorFlow that is expected to be integrated into a\n  custom user-specified binary.\n\n  To include Trainer in a TFX pipeline, configure your pipeline similar to\n  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_simple.py#L104.\n\n  For more details on the Trainer component itself, please refer to\n  https://tensorflow.org/tfx/guide/trainer.  For a tutorial on TF Estimator,\n  please refer to https://www.tensorflow.org/extend/estimators.\n\n  How to create a trainer callback function to be used by this Trainer executor:\n  An estimator can be executed by TFX by first creating a trainer_fn callback\n  method that returns an estimator and some additional parameters, similar to\n  https://github.com/tensorflow/tfx/blob/master/tfx/examples/chicago_taxi_pipeline/taxi_utils.py#L285.\n  This becomes the basis of the new Executor for Trainer. This Executor will\n  then train and evaluate this estimator using the\n  tf.estimator.train_and_evaluate API to train locally.\n  \"\"\"\n\n  # Name of subdirectory which contains checkpoints from prior runs\n  _CHECKPOINT_FILE_NAME = 'checkpoint'\n\n  def _GetTrainerFn(self, exec_properties: Dict[Text, Any]) -> Any:\n    \"\"\"Loads and returns user-defined trainer_fn.\"\"\"\n\n    has_module_file = bool(exec_properties.get('module_file'))\n    has_trainer_fn = bool(exec_properties.get('trainer_fn'))\n\n    if has_module_file == has_trainer_fn:\n      raise ValueError(\n          \"Neither or both of 'module_file' 'trainer_fn' have been supplied in \"\n          \"'exec_properties'.\")\n\n    if has_module_file:\n      return import_utils.import_func_from_source(\n          exec_properties['module_file'], 'trainer_fn')\n\n    trainer_fn_path_split = exec_properties['trainer_fn'].split('.')\n    return import_utils.import_func_from_module(\n        '.'.join(trainer_fn_path_split[0:-1]), trainer_fn_path_split[-1])\n\n  def Do(self, input_dict: Dict[Text, List[types.Artifact]],\n         output_dict: Dict[Text, List[types.Artifact]],\n         exec_properties: Dict[Text, Any]) -> None:\n    \"\"\"Uses a user-supplied tf.estimator to train a TensorFlow model locally.\n\n    The Trainer Executor invokes a training_fn callback function provided by\n    the user via the module_file parameter.  With the tf.estimator returned by\n    this function, the Trainer Executor then builds a TensorFlow model using the\n    user-provided tf.estimator.\n\n    Args:\n      input_dict: Input dict from input key to a list of ML-Metadata Artifacts.\n        - examples: Examples used for training, must include 'train' and 'eval'\n          splits.\n        - transform_output: Optional input transform graph.\n        - schema: Schema of the data.\n      output_dict: Output dict from output key to a list of Artifacts.\n        - output: Exported model.\n      exec_properties: A dict of execution properties.\n        - train_args: JSON string of trainer_pb2.TrainArgs instance, providing\n          args for training.\n        - eval_args: JSON string of trainer_pb2.EvalArgs instance, providing\n          args for eval.\n        - module_file: Python module file containing UDF model definition.\n        - warm_starting: Whether or not we need to do warm starting.\n        - warm_start_from: Optional. If warm_starting is True, this is the\n          directory to find previous model to warm start on.\n\n    Returns:\n      None\n\n    Raises:\n      ValueError: When neither or both of 'module_file' and 'trainer_fn'\n        are present in 'exec_properties'.\n    \"\"\"\n    self._log_startup(input_dict, output_dict, exec_properties)\n\n    # TODO(zhitaoli): Deprecate this in a future version.\n    if exec_properties.get('custom_config', None):\n      cmle_args = exec_properties.get('custom_config',\n                                      {}).get('cmle_training_args')\n      if cmle_args:\n        executor_class_path = '.'.join([Executor.__module__, Executor.__name__])\n        tf.logging.warn(\n            'Passing \\'cmle_training_args\\' to trainer directly is deprecated, '\n            'please use extension executor at '\n            'tfx.extensions.google_cloud_ai_platform.trainer.executor instead')\n\n        return runner.start_cmle_training(input_dict, output_dict,\n                                          exec_properties, executor_class_path,\n                                          cmle_args)\n\n    trainer_fn = self._GetTrainerFn(exec_properties)\n\n    # Set up training parameters\n    train_files = [\n        _all_files_pattern(\n            artifact_utils.get_split_uri(input_dict['examples'], 'train'))\n    ]\n    transform_output = artifact_utils.get_single_uri(\n        input_dict['transform_output']) if input_dict.get(\n            'transform_output', None) else None\n    eval_files = [\n        _all_files_pattern(\n            artifact_utils.get_split_uri(input_dict['examples'], 'eval'))\n    ]\n    schema_file = io_utils.get_only_uri_in_dir(\n        artifact_utils.get_single_uri(input_dict['schema']))\n\n    train_args = trainer_pb2.TrainArgs()\n    eval_args = trainer_pb2.EvalArgs()\n    json_format.Parse(exec_properties['train_args'], train_args)\n    json_format.Parse(exec_properties['eval_args'], eval_args)\n\n    # https://github.com/tensorflow/tfx/issues/45: Replace num_steps=0 with\n    # num_steps=None.  Conversion of the proto to python will set the default\n    # value of an int as 0 so modify the value here.  Tensorflow will raise an\n    # error if num_steps <= 0.\n    train_steps = train_args.num_steps or None\n    eval_steps = eval_args.num_steps or None\n\n    output_path = artifact_utils.get_single_uri(output_dict['output'])\n    serving_model_dir = path_utils.serving_model_dir(output_path)\n    eval_model_dir = path_utils.eval_model_dir(output_path)\n\n    # Assemble warm start path if needed.\n    warm_start_from = None\n    if exec_properties.get('warm_starting') and exec_properties.get(\n        'warm_start_from'):\n      previous_model_dir = os.path.join(exec_properties['warm_start_from'],\n                                        path_utils.SERVING_MODEL_DIR)\n      if previous_model_dir and tf.gfile.Exists(\n          os.path.join(previous_model_dir, self._CHECKPOINT_FILE_NAME)):\n        warm_start_from = previous_model_dir\n\n    # TODO(b/126242806) Use PipelineInputs when it is available in third_party.\n    hparams = tf.contrib.training.HParams(\n        # A list of uris for train files.\n        train_files=train_files,\n        # An optional single uri for transform graph produced by TFT. Will be\n        # None if not specified.\n        transform_output=transform_output,\n        # A single uri for the output directory of the serving model.\n        serving_model_dir=serving_model_dir,\n        # A list of uris for eval files.\n        eval_files=eval_files,\n        # A single uri for schema file.\n        schema_file=schema_file,\n        # Number of train steps.\n        train_steps=train_steps,\n        # Number of eval steps.\n        eval_steps=eval_steps,\n        # A single uri for the model directory to warm start from.\n        warm_start_from=warm_start_from)\n\n    schema = io_utils.parse_pbtxt_file(schema_file, schema_pb2.Schema())\n\n    training_spec = trainer_fn(hparams, schema)\n\n    # Train the model\n    tf.logging.info('Training model.')\n    tf.estimator.train_and_evaluate(training_spec['estimator'],\n                                    training_spec['train_spec'],\n                                    training_spec['eval_spec'])\n    tf.logging.info('Training complete.  Model written to %s',\n                    serving_model_dir)\n\n    # Export an eval savedmodel for TFMA\n    tf.logging.info('Exporting eval_savedmodel for TFMA.')\n    tfma.export.export_eval_savedmodel(\n        estimator=training_spec['estimator'],\n        export_dir_base=eval_model_dir,\n        eval_input_receiver_fn=training_spec['eval_input_receiver_fn'])\n\n    tf.logging.info('Exported eval_savedmodel to %s.', eval_model_dir)\n", "path": "tfx/components/trainer/executor.py"}]}