We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
allow output artifact store configuration (vs hard coded)
it seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). 

see: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148

it would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.
i suggest making it configurable, i can do such PR if we agree its needed. 
flexible pipeline service (host) path in client SDK 
when creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:

`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`

to:

`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`

also note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug

if its acceptable i can submit a PR for the line change above
 

--- 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/iris/iris.py`
Content:
```
1 #!/usr/bin/env python3
2 # Copyright 2020 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 """Iris flowers example using TFX. Based on https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_pipeline_native_keras.py"""
16 
17 from __future__ import absolute_import
18 from __future__ import division
19 from __future__ import print_function
20 
21 import os
22 import kfp
23 from typing import Text
24 
25 import absl
26 import tensorflow_model_analysis as tfma
27 
28 from tfx.components import CsvExampleGen
29 from tfx.components import Evaluator
30 from tfx.components import ExampleValidator
31 from tfx.components import Pusher
32 from tfx.components import ResolverNode
33 from tfx.components import SchemaGen
34 from tfx.components import StatisticsGen
35 from tfx.components import Trainer
36 from tfx.components import Transform
37 from tfx.components.base import executor_spec
38 from tfx.components.trainer.executor import GenericExecutor
39 from tfx.dsl.experimental import latest_blessed_model_resolver
40 from tfx.orchestration import data_types
41 from tfx.orchestration import pipeline
42 from tfx.orchestration.kubeflow import kubeflow_dag_runner
43 from tfx.proto import trainer_pb2
44 from tfx.proto import pusher_pb2
45 from tfx.types import Channel
46 from tfx.types.standard_artifacts import Model
47 from tfx.types.standard_artifacts import ModelBlessing
48 from tfx.utils.dsl_utils import external_input
49 
50 _pipeline_name = 'iris_native_keras'
51 
52 # This example assumes that Iris flowers data is stored in GCS and the
53 # utility function is in iris_utils.py. Feel free to customize as needed.
54 _data_root_param = data_types.RuntimeParameter(
55     name='data-root',
56     default='gs://ml-pipeline/sample-data/iris/data',
57     ptype=Text,
58 )
59 
60 # Python module file to inject customized logic into the TFX components. The
61 # Transform and Trainer both require user-defined functions to run successfully.
62 # This file is fork from https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_utils_native_keras.py
63 # and baked into the TFX image used in the pipeline.
64 _module_file_param = data_types.RuntimeParameter(
65     name='module-file',
66     default=
67     '/tfx-src/tfx/examples/iris/iris_utils_native_keras.py',
68     ptype=Text,
69 )
70 
71 # Directory and data locations. This example assumes all of the flowers
72 # example code and metadata library is relative to a GCS path.
73 # Note: if one deployed KFP from GKE marketplace, it's possible to leverage
74 # the following magic placeholder to auto-populate the default GCS bucket
75 # associated with KFP deployment. Otherwise you'll need to replace it with your
76 # actual bucket name here or when creating a run.
77 _pipeline_root = os.path.join(
78     'gs://{{kfp-default-bucket}}', 'tfx_iris', kfp.dsl.RUN_ID_PLACEHOLDER
79 )
80 
81 
82 def _create_pipeline(
83     pipeline_name: Text, pipeline_root: Text
84 ) -> pipeline.Pipeline:
85   """Implements the Iris flowers pipeline with TFX."""
86   examples = external_input(_data_root_param)
87 
88   # Brings data into the pipeline or otherwise joins/converts training data.
89   example_gen = CsvExampleGen(input=examples)
90 
91   # Computes statistics over data for visualization and example validation.
92   statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])
93 
94   # Generates schema based on statistics files.
95   infer_schema = SchemaGen(
96       statistics=statistics_gen.outputs['statistics'], infer_feature_shape=True
97   )
98 
99   # Performs anomaly detection based on statistics and data schema.
100   validate_stats = ExampleValidator(
101       statistics=statistics_gen.outputs['statistics'],
102       schema=infer_schema.outputs['schema']
103   )
104 
105   # Performs transformations and feature engineering in training and serving.
106   transform = Transform(
107       examples=example_gen.outputs['examples'],
108       schema=infer_schema.outputs['schema'],
109       module_file=_module_file_param
110   )
111 
112   # Uses user-provided Python function that implements a model using Keras.
113   trainer = Trainer(
114       module_file=_module_file_param,
115       custom_executor_spec=executor_spec.ExecutorClassSpec(GenericExecutor),
116       examples=transform.outputs['transformed_examples'],
117       transform_graph=transform.outputs['transform_graph'],
118       schema=infer_schema.outputs['schema'],
119       train_args=trainer_pb2.TrainArgs(num_steps=100),
120       eval_args=trainer_pb2.EvalArgs(num_steps=50)
121   )
122 
123   # Get the latest blessed model for model validation.
124   model_resolver = ResolverNode(
125       instance_name='latest_blessed_model_resolver',
126       resolver_class=latest_blessed_model_resolver.LatestBlessedModelResolver,
127       model=Channel(type=Model),
128       model_blessing=Channel(type=ModelBlessing)
129   )
130 
131   # Uses TFMA to compute an evaluation statistics over features of a model and
132   # perform quality validation of a candidate model (compared to a baseline).
133   # Note: to compile this successfully you'll need TFMA at >= 0.21.5
134   eval_config = tfma.EvalConfig(
135       model_specs=[
136           tfma.ModelSpec(name='candidate', label_key='variety'),
137           tfma.ModelSpec(
138               name='baseline', label_key='variety', is_baseline=True
139           )
140       ],
141       slicing_specs=[
142           tfma.SlicingSpec(),
143           # Data can be sliced along a feature column. Required by TFMA visualization.
144           tfma.SlicingSpec(feature_keys=['sepal_length'])],
145       metrics_specs=[
146           tfma.MetricsSpec(
147               metrics=[
148                   tfma.MetricConfig(
149                       class_name='SparseCategoricalAccuracy',
150                       threshold=tfma.config.MetricThreshold(
151                           value_threshold=tfma.GenericValueThreshold(
152                               lower_bound={'value': 0.9}
153                           ),
154                           change_threshold=tfma.GenericChangeThreshold(
155                               direction=tfma.MetricDirection.HIGHER_IS_BETTER,
156                               absolute={'value': -1e-10}
157                           )
158                       )
159                   )
160               ]
161           )
162       ]
163   )
164 
165   # Uses TFMA to compute a evaluation statistics over features of a model.
166   model_analyzer = Evaluator(
167       examples=example_gen.outputs['examples'],
168       model=trainer.outputs['model'],
169       baseline_model=model_resolver.outputs['model'],
170       # Change threshold will be ignored if there is no baseline (first run).
171       eval_config=eval_config
172   )
173 
174   # Checks whether the model passed the validation steps and pushes the model
175   # to a file destination if check passed.
176   pusher = Pusher(
177       model=trainer.outputs['model'],
178       model_blessing=model_analyzer.outputs['blessing'],
179       push_destination=pusher_pb2.PushDestination(
180           filesystem=pusher_pb2.PushDestination.Filesystem(
181               base_directory=os.path.
182               join(str(pipeline.ROOT_PARAMETER), 'model_serving')
183           )
184       )
185   )
186 
187   return pipeline.Pipeline(
188       pipeline_name=pipeline_name,
189       pipeline_root=pipeline_root,
190       components=[
191           example_gen, statistics_gen, infer_schema, validate_stats, transform,
192           trainer, model_resolver, model_analyzer, pusher
193       ],
194       enable_cache=True,
195   )
196 
197 
198 if __name__ == '__main__':
199   absl.logging.set_verbosity(absl.logging.INFO)
200   # Make sure the version of TFX image used is consistent with the version of
201   # TFX SDK. Here we use tfx:0.22.0 image.
202   config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
203       kubeflow_metadata_config=kubeflow_dag_runner.
204       get_default_kubeflow_metadata_config(),
205       tfx_image='gcr.io/tfx-oss-public/tfx:0.22.0',
206   )
207   kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(
208       output_filename=__file__ + '.yaml', config=config
209   )
210   kfp_runner.run(
211       _create_pipeline(
212           pipeline_name=_pipeline_name, pipeline_root=_pipeline_root
213       )
214   )
215
```
--- 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/iris/iris.py b/samples/core/iris/iris.py
--- a/samples/core/iris/iris.py
+++ b/samples/core/iris/iris.py
@@ -14,10 +14,6 @@
 # limitations under the License.
 """Iris flowers example using TFX. Based on https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_pipeline_native_keras.py"""
 
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
 import os
 import kfp
 from typing import Text

{"golden_diff": "diff --git a/samples/core/iris/iris.py b/samples/core/iris/iris.py\n--- a/samples/core/iris/iris.py\n+++ b/samples/core/iris/iris.py\n@@ -14,10 +14,6 @@\n # limitations under the License.\n \"\"\"Iris flowers example using TFX. Based on https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_pipeline_native_keras.py\"\"\"\n \n-from __future__ import absolute_import\n-from __future__ import division\n-from __future__ import print_function\n-\n import os\n import kfp\n from typing import Text\n", "issue": "allow output artifact store configuration (vs hard coded)\nit seems like the output artifacts are always stored in a specific minio service, port, namespace, bucket, secrets, etc (`minio-service.kubeflow:9000`). \r\n\r\nsee: https://github.com/kubeflow/pipelines/blob/f40a22a3f4a8e06d20cf3e3f425b5058d5c87e0b/sdk/python/kfp/compiler/_op_to_template.py#L148\r\n\r\nit would be great to make it flexible, e.g. allow using S3, or change namespace or bucket names.\r\ni suggest making it configurable, i can do such PR if we agree its needed. \nflexible pipeline service (host) path in client SDK \nwhen creating an SDK `Client()` the path to `ml-pipeline` API service is loaded from a hard coded value (`ml-pipeline.kubeflow.svc.cluster.local:8888`) which indicate a specific k8s namespace. it can be valuable to load that default value from an env variable, i.e. changing the line in `_client.py` from:\r\n\r\n`config.host = host if host else Client.IN_CLUSTER_DNS_NAME`\r\n\r\nto:\r\n\r\n`config.host = host or os.environ.get('ML_PIPELINE_DNS_NAME',Client.IN_CLUSTER_DNS_NAME)`\r\n\r\nalso note that when a user provide the `host` parameter, the ipython output points to the API server and not to the UI service (see the logic in `_get_url_prefix()`), it seems like a potential bug\r\n\r\nif its acceptable i can submit a PR for the line change above\r\n \n", "before_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2020 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\"\"\"Iris flowers example using TFX. Based on https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_pipeline_native_keras.py\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport os\nimport kfp\nfrom typing import Text\n\nimport absl\nimport tensorflow_model_analysis as tfma\n\nfrom tfx.components import CsvExampleGen\nfrom tfx.components import Evaluator\nfrom tfx.components import ExampleValidator\nfrom tfx.components import Pusher\nfrom tfx.components import ResolverNode\nfrom tfx.components import SchemaGen\nfrom tfx.components import StatisticsGen\nfrom tfx.components import Trainer\nfrom tfx.components import Transform\nfrom tfx.components.base import executor_spec\nfrom tfx.components.trainer.executor import GenericExecutor\nfrom tfx.dsl.experimental import latest_blessed_model_resolver\nfrom tfx.orchestration import data_types\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.proto import trainer_pb2\nfrom tfx.proto import pusher_pb2\nfrom tfx.types import Channel\nfrom tfx.types.standard_artifacts import Model\nfrom tfx.types.standard_artifacts import ModelBlessing\nfrom tfx.utils.dsl_utils import external_input\n\n_pipeline_name = 'iris_native_keras'\n\n# This example assumes that Iris flowers data is stored in GCS and the\n# utility function is in iris_utils.py. Feel free to customize as needed.\n_data_root_param = data_types.RuntimeParameter(\n    name='data-root',\n    default='gs://ml-pipeline/sample-data/iris/data',\n    ptype=Text,\n)\n\n# Python module file to inject customized logic into the TFX components. The\n# Transform and Trainer both require user-defined functions to run successfully.\n# This file is fork from https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_utils_native_keras.py\n# and baked into the TFX image used in the pipeline.\n_module_file_param = data_types.RuntimeParameter(\n    name='module-file',\n    default=\n    '/tfx-src/tfx/examples/iris/iris_utils_native_keras.py',\n    ptype=Text,\n)\n\n# Directory and data locations. This example assumes all of the flowers\n# example code and metadata library is relative to a GCS path.\n# Note: if one deployed KFP from GKE marketplace, it's possible to leverage\n# the following magic placeholder to auto-populate the default GCS bucket\n# associated with KFP deployment. Otherwise you'll need to replace it with your\n# actual bucket name here or when creating a run.\n_pipeline_root = os.path.join(\n    'gs://{{kfp-default-bucket}}', 'tfx_iris', kfp.dsl.RUN_ID_PLACEHOLDER\n)\n\n\ndef _create_pipeline(\n    pipeline_name: Text, pipeline_root: Text\n) -> pipeline.Pipeline:\n  \"\"\"Implements the Iris flowers pipeline with TFX.\"\"\"\n  examples = external_input(_data_root_param)\n\n  # Brings data into the pipeline or otherwise joins/converts training data.\n  example_gen = CsvExampleGen(input=examples)\n\n  # Computes statistics over data for visualization and example validation.\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n\n  # Generates schema based on statistics files.\n  infer_schema = SchemaGen(\n      statistics=statistics_gen.outputs['statistics'], infer_feature_shape=True\n  )\n\n  # Performs anomaly detection based on statistics and data schema.\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema']\n  )\n\n  # Performs transformations and feature engineering in training and serving.\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      module_file=_module_file_param\n  )\n\n  # Uses user-provided Python function that implements a model using Keras.\n  trainer = Trainer(\n      module_file=_module_file_param,\n      custom_executor_spec=executor_spec.ExecutorClassSpec(GenericExecutor),\n      examples=transform.outputs['transformed_examples'],\n      transform_graph=transform.outputs['transform_graph'],\n      schema=infer_schema.outputs['schema'],\n      train_args=trainer_pb2.TrainArgs(num_steps=100),\n      eval_args=trainer_pb2.EvalArgs(num_steps=50)\n  )\n\n  # Get the latest blessed model for model validation.\n  model_resolver = ResolverNode(\n      instance_name='latest_blessed_model_resolver',\n      resolver_class=latest_blessed_model_resolver.LatestBlessedModelResolver,\n      model=Channel(type=Model),\n      model_blessing=Channel(type=ModelBlessing)\n  )\n\n  # Uses TFMA to compute an evaluation statistics over features of a model and\n  # perform quality validation of a candidate model (compared to a baseline).\n  # Note: to compile this successfully you'll need TFMA at >= 0.21.5\n  eval_config = tfma.EvalConfig(\n      model_specs=[\n          tfma.ModelSpec(name='candidate', label_key='variety'),\n          tfma.ModelSpec(\n              name='baseline', label_key='variety', is_baseline=True\n          )\n      ],\n      slicing_specs=[\n          tfma.SlicingSpec(),\n          # Data can be sliced along a feature column. Required by TFMA visualization.\n          tfma.SlicingSpec(feature_keys=['sepal_length'])],\n      metrics_specs=[\n          tfma.MetricsSpec(\n              metrics=[\n                  tfma.MetricConfig(\n                      class_name='SparseCategoricalAccuracy',\n                      threshold=tfma.config.MetricThreshold(\n                          value_threshold=tfma.GenericValueThreshold(\n                              lower_bound={'value': 0.9}\n                          ),\n                          change_threshold=tfma.GenericChangeThreshold(\n                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,\n                              absolute={'value': -1e-10}\n                          )\n                      )\n                  )\n              ]\n          )\n      ]\n  )\n\n  # Uses TFMA to compute a evaluation statistics over features of a model.\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model=trainer.outputs['model'],\n      baseline_model=model_resolver.outputs['model'],\n      # Change threshold will be ignored if there is no baseline (first run).\n      eval_config=eval_config\n  )\n\n  # Checks whether the model passed the validation steps and pushes the model\n  # to a file destination if check passed.\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=pipeline_name,\n      pipeline_root=pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_resolver, model_analyzer, pusher\n      ],\n      enable_cache=True,\n  )\n\n\nif __name__ == '__main__':\n  absl.logging.set_verbosity(absl.logging.INFO)\n  # Make sure the version of TFX image used is consistent with the version of\n  # TFX SDK. Here we use tfx:0.22.0 image.\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  kfp_runner.run(\n      _create_pipeline(\n          pipeline_name=_pipeline_name, pipeline_root=_pipeline_root\n      )\n  )\n", "path": "samples/core/iris/iris.py"}], "after_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2020 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\"\"\"Iris flowers example using TFX. Based on https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_pipeline_native_keras.py\"\"\"\n\nimport os\nimport kfp\nfrom typing import Text\n\nimport absl\nimport tensorflow_model_analysis as tfma\n\nfrom tfx.components import CsvExampleGen\nfrom tfx.components import Evaluator\nfrom tfx.components import ExampleValidator\nfrom tfx.components import Pusher\nfrom tfx.components import ResolverNode\nfrom tfx.components import SchemaGen\nfrom tfx.components import StatisticsGen\nfrom tfx.components import Trainer\nfrom tfx.components import Transform\nfrom tfx.components.base import executor_spec\nfrom tfx.components.trainer.executor import GenericExecutor\nfrom tfx.dsl.experimental import latest_blessed_model_resolver\nfrom tfx.orchestration import data_types\nfrom tfx.orchestration import pipeline\nfrom tfx.orchestration.kubeflow import kubeflow_dag_runner\nfrom tfx.proto import trainer_pb2\nfrom tfx.proto import pusher_pb2\nfrom tfx.types import Channel\nfrom tfx.types.standard_artifacts import Model\nfrom tfx.types.standard_artifacts import ModelBlessing\nfrom tfx.utils.dsl_utils import external_input\n\n_pipeline_name = 'iris_native_keras'\n\n# This example assumes that Iris flowers data is stored in GCS and the\n# utility function is in iris_utils.py. Feel free to customize as needed.\n_data_root_param = data_types.RuntimeParameter(\n    name='data-root',\n    default='gs://ml-pipeline/sample-data/iris/data',\n    ptype=Text,\n)\n\n# Python module file to inject customized logic into the TFX components. The\n# Transform and Trainer both require user-defined functions to run successfully.\n# This file is fork from https://github.com/tensorflow/tfx/blob/master/tfx/examples/iris/iris_utils_native_keras.py\n# and baked into the TFX image used in the pipeline.\n_module_file_param = data_types.RuntimeParameter(\n    name='module-file',\n    default=\n    '/tfx-src/tfx/examples/iris/iris_utils_native_keras.py',\n    ptype=Text,\n)\n\n# Directory and data locations. This example assumes all of the flowers\n# example code and metadata library is relative to a GCS path.\n# Note: if one deployed KFP from GKE marketplace, it's possible to leverage\n# the following magic placeholder to auto-populate the default GCS bucket\n# associated with KFP deployment. Otherwise you'll need to replace it with your\n# actual bucket name here or when creating a run.\n_pipeline_root = os.path.join(\n    'gs://{{kfp-default-bucket}}', 'tfx_iris', kfp.dsl.RUN_ID_PLACEHOLDER\n)\n\n\ndef _create_pipeline(\n    pipeline_name: Text, pipeline_root: Text\n) -> pipeline.Pipeline:\n  \"\"\"Implements the Iris flowers pipeline with TFX.\"\"\"\n  examples = external_input(_data_root_param)\n\n  # Brings data into the pipeline or otherwise joins/converts training data.\n  example_gen = CsvExampleGen(input=examples)\n\n  # Computes statistics over data for visualization and example validation.\n  statistics_gen = StatisticsGen(examples=example_gen.outputs['examples'])\n\n  # Generates schema based on statistics files.\n  infer_schema = SchemaGen(\n      statistics=statistics_gen.outputs['statistics'], infer_feature_shape=True\n  )\n\n  # Performs anomaly detection based on statistics and data schema.\n  validate_stats = ExampleValidator(\n      statistics=statistics_gen.outputs['statistics'],\n      schema=infer_schema.outputs['schema']\n  )\n\n  # Performs transformations and feature engineering in training and serving.\n  transform = Transform(\n      examples=example_gen.outputs['examples'],\n      schema=infer_schema.outputs['schema'],\n      module_file=_module_file_param\n  )\n\n  # Uses user-provided Python function that implements a model using Keras.\n  trainer = Trainer(\n      module_file=_module_file_param,\n      custom_executor_spec=executor_spec.ExecutorClassSpec(GenericExecutor),\n      examples=transform.outputs['transformed_examples'],\n      transform_graph=transform.outputs['transform_graph'],\n      schema=infer_schema.outputs['schema'],\n      train_args=trainer_pb2.TrainArgs(num_steps=100),\n      eval_args=trainer_pb2.EvalArgs(num_steps=50)\n  )\n\n  # Get the latest blessed model for model validation.\n  model_resolver = ResolverNode(\n      instance_name='latest_blessed_model_resolver',\n      resolver_class=latest_blessed_model_resolver.LatestBlessedModelResolver,\n      model=Channel(type=Model),\n      model_blessing=Channel(type=ModelBlessing)\n  )\n\n  # Uses TFMA to compute an evaluation statistics over features of a model and\n  # perform quality validation of a candidate model (compared to a baseline).\n  # Note: to compile this successfully you'll need TFMA at >= 0.21.5\n  eval_config = tfma.EvalConfig(\n      model_specs=[\n          tfma.ModelSpec(name='candidate', label_key='variety'),\n          tfma.ModelSpec(\n              name='baseline', label_key='variety', is_baseline=True\n          )\n      ],\n      slicing_specs=[\n          tfma.SlicingSpec(),\n          # Data can be sliced along a feature column. Required by TFMA visualization.\n          tfma.SlicingSpec(feature_keys=['sepal_length'])],\n      metrics_specs=[\n          tfma.MetricsSpec(\n              metrics=[\n                  tfma.MetricConfig(\n                      class_name='SparseCategoricalAccuracy',\n                      threshold=tfma.config.MetricThreshold(\n                          value_threshold=tfma.GenericValueThreshold(\n                              lower_bound={'value': 0.9}\n                          ),\n                          change_threshold=tfma.GenericChangeThreshold(\n                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,\n                              absolute={'value': -1e-10}\n                          )\n                      )\n                  )\n              ]\n          )\n      ]\n  )\n\n  # Uses TFMA to compute a evaluation statistics over features of a model.\n  model_analyzer = Evaluator(\n      examples=example_gen.outputs['examples'],\n      model=trainer.outputs['model'],\n      baseline_model=model_resolver.outputs['model'],\n      # Change threshold will be ignored if there is no baseline (first run).\n      eval_config=eval_config\n  )\n\n  # Checks whether the model passed the validation steps and pushes the model\n  # to a file destination if check passed.\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=pipeline_name,\n      pipeline_root=pipeline_root,\n      components=[\n          example_gen, statistics_gen, infer_schema, validate_stats, transform,\n          trainer, model_resolver, model_analyzer, pusher\n      ],\n      enable_cache=True,\n  )\n\n\nif __name__ == '__main__':\n  absl.logging.set_verbosity(absl.logging.INFO)\n  # Make sure the version of TFX image used is consistent with the version of\n  # TFX SDK. Here we use tfx:0.21.2 image.\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.21.2',\n  )\n  kfp_runner = kubeflow_dag_runner.KubeflowDagRunner(\n      output_filename=__file__ + '.yaml', config=config\n  )\n  kfp_runner.run(\n      _create_pipeline(\n          pipeline_name=_pipeline_name, pipeline_root=_pipeline_root\n      )\n  )\n", "path": "samples/core/iris/iris.py"}]}