We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Problems upgrading to TFX 0.27.0
### What steps did you take:
Installed Kubeflow Pipelines on GCP via kustomize manifests.
Tried to run the Taxi TFX Demo.

### What happened:
On the first step, I got the error "No module named 'tfx.dsl.components'"

### What did you expect to happen:
To successfully run the TFX Taxi Demo.

### Environment:
How did you deploy Kubeflow Pipelines (KFP)?
Via the kustomize manifests in GCP.

KFP version: 1.4.0-rc.1

/kind bug
/area backend

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py`
Content:
```
1 #!/usr/bin/env python3
2 # Copyright 2019 Google LLC
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #      http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import os
17 
18 from typing import Text
19 
20 import kfp
21 import tensorflow_model_analysis as tfma
22 from tfx.components.evaluator.component import Evaluator
23 from tfx.components.example_gen.csv_example_gen.component import CsvExampleGen
24 from tfx.components.example_validator.component import ExampleValidator
25 from tfx.components.pusher.component import Pusher
26 from tfx.components.schema_gen.component import SchemaGen
27 from tfx.components.statistics_gen.component import StatisticsGen
28 from tfx.components.trainer.component import Trainer
29 from tfx.components.transform.component import Transform
30 from tfx.orchestration import data_types
31 from tfx.orchestration import pipeline
32 from tfx.orchestration.kubeflow import kubeflow_dag_runner
33 from tfx.utils.dsl_utils import external_input
34 from tfx.proto import pusher_pb2
35 from tfx.proto import trainer_pb2
36 
37 # Define pipeline params used for pipeline execution.
38 # Path to the module file, should be a GCS path,
39 # or a module file baked in the docker image used by the pipeline.
40 _taxi_module_file_param = data_types.RuntimeParameter(
41     name='module-file',
42     default='/tfx-src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',
43     ptype=Text,
44 )
45 
46 # Path to the CSV data file, under which their should be a data.csv file.
47 _data_root_param = data_types.RuntimeParameter(
48     name='data-root',
49     default='gs://ml-pipeline/sample-data/chicago-taxi/data',
50     ptype=Text,
51 )
52 
53 # Path of pipeline root, should be a GCS path.
54 pipeline_root = os.path.join(
55     'gs://{{kfp-default-bucket}}', 'tfx_taxi_simple', kfp.dsl.RUN_ID_PLACEHOLDER
56 )
57 
58 
59 def _create_pipeline(
60     pipeline_root: Text, csv_input_location: data_types.RuntimeParameter,
61     taxi_module_file: data_types.RuntimeParameter, enable_cache: bool
62 ):
63   """Creates a simple Kubeflow-based Chicago Taxi TFX pipeline.
64 
65   Args:
66     pipeline_root: The root of the pipeline output.
67     csv_input_location: The location of the input data directory.
68     taxi_module_file: The location of the module file for Transform/Trainer.
69     enable_cache: Whether to enable cache or not.
70 
71   Returns:
72     A logical TFX pipeline.Pipeline object.
73   """
74   examples = external_input(csv_input_location)
75 
76   example_gen = CsvExampleGen(input=examples)
77   statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])
78   infer_schema = SchemaGen(
79       statistics=statistics_gen.outputs['statistics'],
80       infer_feature_shape=False,
81   )
82   validate_stats = ExampleValidator(
83       statistics=statistics_gen.outputs['statistics'],
84       schema=infer_schema.outputs['schema'],
85   )
86   transform = Transform(
87       examples=example_gen.outputs['examples'],
88       schema=infer_schema.outputs['schema'],
89       module_file=taxi_module_file,
90   )
91   trainer = Trainer(
92       module_file=taxi_module_file,
93       transformed_examples=transform.outputs['transformed_examples'],
94       schema=infer_schema.outputs['schema'],
95       transform_graph=transform.outputs['transform_graph'],
96       train_args=trainer_pb2.TrainArgs(num_steps=10),
97       eval_args=trainer_pb2.EvalArgs(num_steps=5),
98   )
99   # Set the TFMA config for Model Evaluation and Validation.
100   eval_config = tfma.EvalConfig(
101       model_specs=[
102           # Using signature 'eval' implies the use of an EvalSavedModel. To use
103           # a serving model remove the signature to defaults to 'serving_default'
104           # and add a label_key.
105           tfma.ModelSpec(signature_name='eval')
106       ],
107       metrics_specs=[
108           tfma.MetricsSpec(
109               # The metrics added here are in addition to those saved with the
110               # model (assuming either a keras model or EvalSavedModel is used).
111               # Any metrics added into the saved model (for example using
112               # model.compile(..., metrics=[...]), etc) will be computed
113               # automatically.
114               metrics=[tfma.MetricConfig(class_name='ExampleCount')],
115               # To add validation thresholds for metrics saved with the model,
116               # add them keyed by metric name to the thresholds map.
117               thresholds={
118                   'binary_accuracy':
119                       tfma.MetricThreshold(
120                           value_threshold=tfma.GenericValueThreshold(
121                               lower_bound={'value': 0.5}
122                           ),
123                           change_threshold=tfma.GenericChangeThreshold(
124                               direction=tfma.MetricDirection.HIGHER_IS_BETTER,
125                               absolute={'value': -1e-10}
126                           )
127                       )
128               }
129           )
130       ],
131       slicing_specs=[
132           # An empty slice spec means the overall slice, i.e. the whole dataset.
133           tfma.SlicingSpec(),
134           # Data can be sliced along a feature column. In this case, data is
135           # sliced along feature column trip_start_hour.
136           tfma.SlicingSpec(feature_keys=['trip_start_hour'])
137       ]
138   )
139 
140   model_analyzer = Evaluator(
141       examples=example_gen.outputs['examples'],
142       model=trainer.outputs['model'],
143       eval_config=eval_config,
144   )
145 
146   pusher = Pusher(
147       model=trainer.outputs['model'],
148       model_blessing=model_analyzer.outputs['blessing'],
149       push_destination=pusher_pb2.PushDestination(
150           filesystem=pusher_pb2.PushDestination.Filesystem(
151               base_directory=os.path.
152               join(str(pipeline.ROOT_PARAMETER), 'model_serving')
153           )
154       ),
155   )
156 
157   return pipeline.Pipeline(
158       pipeline_name='parameterized_tfx_oss',
159       pipeline_root=pipeline_root,
160       components=[
161           example_gen, statistics_gen, infer_schema, validate_stats, transform,
162           trainer, model_analyzer, pusher
163       ],
164       enable_cache=enable_cache,
165   )
166 
167 
168 if __name__ == '__main__':
169   enable_cache = True
170   pipeline = _create_pipeline(
171       pipeline_root,
172       _data_root_param,
173       _taxi_module_file_param,
174       enable_cache=enable_cache,
175   )
176   # Make sure the version of TFX image used is consistent with the version of
177   # TFX SDK.
178   config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
179       kubeflow_metadata_config=kubeflow_dag_runner.
180       get_default_kubeflow_metadata_config(),
181       tfx_image='gcr.io/tfx-oss-public/tfx:0.22.0',
182   )
183   kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(
184       output_filename=__file__ + '.yaml', config=config
185   )
186 
187   kfp_runner.run(pipeline)
188
```
--- 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/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py b/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py
--- a/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py
+++ b/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py
@@ -39,7 +39,7 @@
 # or a module file baked in the docker image used by the pipeline.
 _taxi_module_file_param = data_types.RuntimeParameter(
     name='module-file',
-    default='/tfx-src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',
+    default='/tfx/src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',
     ptype=Text,
 )
 
@@ -178,7 +178,7 @@
   config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
       kubeflow_metadata_config=kubeflow_dag_runner.
       get_default_kubeflow_metadata_config(),
-      tfx_image='gcr.io/tfx-oss-public/tfx:0.22.0',
+      tfx_image='gcr.io/tfx-oss-public/tfx:0.27.0',
   )
   kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(
       output_filename=__file__ + '.yaml', config=config

{"golden_diff": "diff --git a/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py b/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py\n--- a/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py\n+++ b/samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py\n@@ -39,7 +39,7 @@\n # or a module file baked in the docker image used by the pipeline.\n _taxi_module_file_param = data_types.RuntimeParameter(\n     name='module-file',\n-    default='/tfx-src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',\n+    default='/tfx/src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',\n     ptype=Text,\n )\n \n@@ -178,7 +178,7 @@\n   config = kubeflow_dag_runner.KubeflowDagRunnerConfig(\n       kubeflow_metadata_config=kubeflow_dag_runner.\n       get_default_kubeflow_metadata_config(),\n-      tfx_image='gcr.io/tfx-oss-public/tfx:0.22.0',\n+      tfx_image='gcr.io/tfx-oss-public/tfx:0.27.0',\n   )\n   kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(\n       output_filename=__file__ + '.yaml', config=config\n", "issue": "Problems upgrading to TFX 0.27.0\n### What steps did you take:\r\nInstalled Kubeflow Pipelines on GCP via kustomize manifests.\r\nTried to run the Taxi TFX Demo.\r\n\r\n### What happened:\r\nOn the first step, I got the error \"No module named 'tfx.dsl.components'\"\r\n\r\n### What did you expect to happen:\r\nTo successfully run the TFX Taxi Demo.\r\n\r\n### Environment:\r\nHow did you deploy Kubeflow Pipelines (KFP)?\r\nVia the kustomize manifests in GCP.\r\n\r\nKFP version: 1.4.0-rc.1\r\n\r\n/kind bug\r\n/area backend\n", "before_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2019 Google LLC\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\nimport os\n\nfrom typing import Text\n\nimport kfp\nimport tensorflow_model_analysis as tfma\nfrom tfx.components.evaluator.component import Evaluator\nfrom tfx.components.example_gen.csv_example_gen.component import CsvExampleGen\nfrom tfx.components.example_validator.component import ExampleValidator\nfrom tfx.components.pusher.component import Pusher\nfrom tfx.components.schema_gen.component import SchemaGen\nfrom tfx.components.statistics_gen.component import StatisticsGen\nfrom tfx.components.trainer.component import Trainer\nfrom tfx.components.transform.component import Transform\nfrom tfx.orchestration import data_types\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.utils.dsl_utils import external_input\nfrom tfx.proto import pusher_pb2\nfrom tfx.proto import trainer_pb2\n\n# Define pipeline params used for pipeline execution.\n# Path to the module file, should be a GCS path,\n# or a module file baked in the docker image used by the pipeline.\n_taxi_module_file_param = data_types.RuntimeParameter(\n    name='module-file',\n    default='/tfx-src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',\n    ptype=Text,\n)\n\n# Path to the CSV data file, under which their should be a data.csv file.\n_data_root_param = data_types.RuntimeParameter(\n    name='data-root',\n    default='gs://ml-pipeline/sample-data/chicago-taxi/data',\n    ptype=Text,\n)\n\n# Path of pipeline root, should be a GCS path.\npipeline_root = os.path.join(\n    'gs://{{kfp-default-bucket}}', 'tfx_taxi_simple', kfp.dsl.RUN_ID_PLACEHOLDER\n)\n\n\ndef _create_pipeline(\n    pipeline_root: Text, csv_input_location: data_types.RuntimeParameter,\n    taxi_module_file: data_types.RuntimeParameter, enable_cache: bool\n):\n  \"\"\"Creates a simple Kubeflow-based Chicago Taxi TFX pipeline.\n\n  Args:\n    pipeline_root: The root of the pipeline output.\n    csv_input_location: The location of the input data directory.\n    taxi_module_file: The location of the module file for Transform/Trainer.\n    enable_cache: Whether to enable cache or not.\n\n  Returns:\n    A logical TFX pipeline.Pipeline object.\n  \"\"\"\n  examples = external_input(csv_input_location)\n\n  example_gen = CsvExampleGen(input=examples)\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n  infer_schema = SchemaGen(\n      statistics=statistics_gen.outputs['statistics'],\n      infer_feature_shape=False,\n  )\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema'],\n  )\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      module_file=taxi_module_file,\n  )\n  trainer = Trainer(\n      module_file=taxi_module_file,\n      transformed_examples=transform.outputs['transformed_examples'],\n      schema=infer_schema.outputs['schema'],\n      transform_graph=transform.outputs['transform_graph'],\n      train_args=trainer_pb2.TrainArgs(num_steps=10),\n      eval_args=trainer_pb2.EvalArgs(num_steps=5),\n  )\n  # Set the TFMA config for Model Evaluation and Validation.\n  eval_config = tfma.EvalConfig(\n      model_specs=[\n          # Using signature 'eval' implies the use of an EvalSavedModel. To use\n          # a serving model remove the signature to defaults to 'serving_default'\n          # and add a label_key.\n          tfma.ModelSpec(signature_name='eval')\n      ],\n      metrics_specs=[\n          tfma.MetricsSpec(\n              # The metrics added here are in addition to those saved with the\n              # model (assuming either a keras model or EvalSavedModel is used).\n              # Any metrics added into the saved model (for example using\n              # model.compile(..., metrics=[...]), etc) will be computed\n              # automatically.\n              metrics=[tfma.MetricConfig(class_name='ExampleCount')],\n              # To add validation thresholds for metrics saved with the model,\n              # add them keyed by metric name to the thresholds map.\n              thresholds={\n                  'binary_accuracy':\n                      tfma.MetricThreshold(\n                          value_threshold=tfma.GenericValueThreshold(\n                              lower_bound={'value': 0.5}\n                          ),\n                          change_threshold=tfma.GenericChangeThreshold(\n                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,\n                              absolute={'value': -1e-10}\n                          )\n                      )\n              }\n          )\n      ],\n      slicing_specs=[\n          # An empty slice spec means the overall slice, i.e. the whole dataset.\n          tfma.SlicingSpec(),\n          # Data can be sliced along a feature column. In this case, data is\n          # sliced along feature column trip_start_hour.\n          tfma.SlicingSpec(feature_keys=['trip_start_hour'])\n      ]\n  )\n\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model=trainer.outputs['model'],\n      eval_config=eval_config,\n  )\n\n  pusher = Pusher(\n      model=trainer.outputs['model'],\n      model_blessing=model_analyzer.outputs['blessing'],\n      push_destination=pusher_pb2.PushDestination(\n          filesystem=pusher_pb2.PushDestination.Filesystem(\n              base_directory=os.path.\n              join(str(pipeline.ROOT_PARAMETER), 'model_serving')\n          )\n      ),\n  )\n\n  return pipeline.Pipeline(\n      pipeline_name='parameterized_tfx_oss',\n      pipeline_root=pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_analyzer, pusher\n      ],\n      enable_cache=enable_cache,\n  )\n\n\nif __name__ == '__main__':\n  enable_cache = True\n  pipeline = _create_pipeline(\n      pipeline_root,\n      _data_root_param,\n      _taxi_module_file_param,\n      enable_cache=enable_cache,\n  )\n  # Make sure the version of TFX image used is consistent with the version of\n  # TFX SDK.\n  config = kubeflow_dag_runner.KubeflowDagRunnerConfig(\n      kubeflow_metadata_config=kubeflow_dag_runner.\n      get_default_kubeflow_metadata_config(),\n      tfx_image='gcr.io/tfx-oss-public/tfx:0.22.0',\n  )\n  kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(\n      output_filename=__file__ + '.yaml', config=config\n  )\n\n  kfp_runner.run(pipeline)\n", "path": "samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py"}], "after_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2019 Google LLC\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\nimport os\n\nfrom typing import Text\n\nimport kfp\nimport tensorflow_model_analysis as tfma\nfrom tfx.components.evaluator.component import Evaluator\nfrom tfx.components.example_gen.csv_example_gen.component import CsvExampleGen\nfrom tfx.components.example_validator.component import ExampleValidator\nfrom tfx.components.pusher.component import Pusher\nfrom tfx.components.schema_gen.component import SchemaGen\nfrom tfx.components.statistics_gen.component import StatisticsGen\nfrom tfx.components.trainer.component import Trainer\nfrom tfx.components.transform.component import Transform\nfrom tfx.orchestration import data_types\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.utils.dsl_utils import external_input\nfrom tfx.proto import pusher_pb2\nfrom tfx.proto import trainer_pb2\n\n# Define pipeline params used for pipeline execution.\n# Path to the module file, should be a GCS path,\n# or a module file baked in the docker image used by the pipeline.\n_taxi_module_file_param = data_types.RuntimeParameter(\n    name='module-file',\n    default='/tfx/src/tfx/examples/chicago_taxi_pipeline/taxi_utils.py',\n    ptype=Text,\n)\n\n# Path to the CSV data file, under which their should be a data.csv file.\n_data_root_param = data_types.RuntimeParameter(\n    name='data-root',\n    default='gs://ml-pipeline/sample-data/chicago-taxi/data',\n    ptype=Text,\n)\n\n# Path of pipeline root, should be a GCS path.\npipeline_root = os.path.join(\n    'gs://{{kfp-default-bucket}}', 'tfx_taxi_simple', kfp.dsl.RUN_ID_PLACEHOLDER\n)\n\n\ndef _create_pipeline(\n    pipeline_root: Text, csv_input_location: data_types.RuntimeParameter,\n    taxi_module_file: data_types.RuntimeParameter, enable_cache: bool\n):\n  \"\"\"Creates a simple Kubeflow-based Chicago Taxi TFX pipeline.\n\n  Args:\n    pipeline_root: The root of the pipeline output.\n    csv_input_location: The location of the input data directory.\n    taxi_module_file: The location of the module file for Transform/Trainer.\n    enable_cache: Whether to enable cache or not.\n\n  Returns:\n    A logical TFX pipeline.Pipeline object.\n  \"\"\"\n  examples = external_input(csv_input_location)\n\n  example_gen = CsvExampleGen(input=examples)\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n  infer_schema = SchemaGen(\n      statistics=statistics_gen.outputs['statistics'],\n      infer_feature_shape=False,\n  )\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema'],\n  )\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      module_file=taxi_module_file,\n  )\n  trainer = Trainer(\n      module_file=taxi_module_file,\n      transformed_examples=transform.outputs['transformed_examples'],\n      schema=infer_schema.outputs['schema'],\n      transform_graph=transform.outputs['transform_graph'],\n      train_args=trainer_pb2.TrainArgs(num_steps=10),\n      eval_args=trainer_pb2.EvalArgs(num_steps=5),\n  )\n  # Set the TFMA config for Model Evaluation and Validation.\n  eval_config = tfma.EvalConfig(\n      model_specs=[\n          # Using signature 'eval' implies the use of an EvalSavedModel. To use\n          # a serving model remove the signature to defaults to 'serving_default'\n          # and add a label_key.\n          tfma.ModelSpec(signature_name='eval')\n      ],\n      metrics_specs=[\n          tfma.MetricsSpec(\n              # The metrics added here are in addition to those saved with the\n              # model (assuming either a keras model or EvalSavedModel is used).\n              # Any metrics added into the saved model (for example using\n              # model.compile(..., metrics=[...]), etc) will be computed\n              # automatically.\n              metrics=[tfma.MetricConfig(class_name='ExampleCount')],\n              # To add validation thresholds for metrics saved with the model,\n              # add them keyed by metric name to the thresholds map.\n              thresholds={\n                  'binary_accuracy':\n                      tfma.MetricThreshold(\n                          value_threshold=tfma.GenericValueThreshold(\n                              lower_bound={'value': 0.5}\n                          ),\n                          change_threshold=tfma.GenericChangeThreshold(\n                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,\n                              absolute={'value': -1e-10}\n                          )\n                      )\n              }\n          )\n      ],\n      slicing_specs=[\n          # An empty slice spec means the overall slice, i.e. the whole dataset.\n          tfma.SlicingSpec(),\n          # Data can be sliced along a feature column. In this case, data is\n          # sliced along feature column trip_start_hour.\n          tfma.SlicingSpec(feature_keys=['trip_start_hour'])\n      ]\n  )\n\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model=trainer.outputs['model'],\n      eval_config=eval_config,\n  )\n\n  pusher = Pusher(\n      model=trainer.outputs['model'],\n      model_blessing=model_analyzer.outputs['blessing'],\n      push_destination=pusher_pb2.PushDestination(\n          filesystem=pusher_pb2.PushDestination.Filesystem(\n              base_directory=os.path.\n              join(str(pipeline.ROOT_PARAMETER), 'model_serving')\n          )\n      ),\n  )\n\n  return pipeline.Pipeline(\n      pipeline_name='parameterized_tfx_oss',\n      pipeline_root=pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_analyzer, pusher\n      ],\n      enable_cache=enable_cache,\n  )\n\n\nif __name__ == '__main__':\n  enable_cache = True\n  pipeline = _create_pipeline(\n      pipeline_root,\n      _data_root_param,\n      _taxi_module_file_param,\n      enable_cache=enable_cache,\n  )\n  # Make sure the version of TFX image used is consistent with the version of\n  # TFX SDK.\n  config = kubeflow_dag_runner.KubeflowDagRunnerConfig(\n      kubeflow_metadata_config=kubeflow_dag_runner.\n      get_default_kubeflow_metadata_config(),\n      tfx_image='gcr.io/tfx-oss-public/tfx:0.27.0',\n  )\n  kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(\n      output_filename=__file__ + '.yaml', config=config\n  )\n\n  kfp_runner.run(pipeline)\n", "path": "samples/core/parameterized_tfx_oss/parameterized_tfx_oss.py"}]}