We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
The KFP preloaded XGboost sample is broken and out-dated. 
**TL;DR: The preload XGBoost sample is currently broken. 
Proposing we remove this sample from KFP preload and from sample test until we got a chance to refresh the sample.**

------------
The direct cause was that it used the Dataproc 1.2 image which is based on Python 2.7, and pip 21.0 dropped support for Python 2.7.
The symptom is that `dataproc_create_cluster` fails on initialization. 
![image](https://user-images.githubusercontent.com/2043310/106823995-bbcd6780-6636-11eb-8169-92d7a338e048.png)
and the specific error is mentioned [here](https://github.com/kubeflow/pipelines/issues/5007#issuecomment-769637030).

https://github.com/kubeflow/pipelines/pull/5062 made an attempted fix by upgrading to Dataproc 1.5 image. It fixed the Dataproc cluster creation issue, but we hit [an error](https://github.com/kubeflow/pipelines/issues/5007#issuecomment-770182313) later at the Trainer step.

We were advised that newer versions of Dataproc images likely don't have XGBoost library preinstalled, as there's now [an initialization action that goes through extra steps to install XGBoost libraries](https://github.com/GoogleCloudDataproc/initialization-actions/tree/master/rapids).

Following that route, I tried installing the default XGBoost version using the rapids script, then hit the error as follows:
```
21/02/03 18:34:20 INFO org.spark_project.jetty.util.log: Logging initialized @3037ms
21/02/03 18:34:20 INFO org.spark_project.jetty.server.Server: jetty-9.3.z-SNAPSHOT, build timestamp: unknown, git hash: unknown
21/02/03 18:34:20 INFO org.spark_project.jetty.server.Server: Started @3169ms
21/02/03 18:34:20 INFO org.spark_project.jetty.server.AbstractConnector: Started ServerConnector@4159e81b{HTTP/1.1,[http/1.1]}{0.0.0.0:37489}
21/02/03 18:34:20 INFO org.apache.hadoop.yarn.client.RMProxy: Connecting to ResourceManager at xgb-bdd8f29b-fb13-4ec2-abcf-38b3699e7ca3-m/10.128.0.101:8032
21/02/03 18:34:21 INFO org.apache.hadoop.yarn.client.AHSProxy: Connecting to Application History server at xgb-bdd8f29b-fb13-4ec2-abcf-38b3699e7ca3-m/10.128.0.101:10200
21/02/03 18:34:21 INFO org.apache.hadoop.conf.Configuration: resource-types.xml not found
21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Unable to find 'resource-types.xml'.
21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Adding resource type - name = memory-mb, units = Mi, type = COUNTABLE
21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Adding resource type - name = vcores, units = , type = COUNTABLE
21/02/03 18:34:23 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl: Submitted application application_1612377093662_0003
21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1
21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1
21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat: DEBUG: Terminated node allocation with : CompletedNodes: 1, size left: 0
21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1
21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1
21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat: DEBUG: Terminated node allocation with : CompletedNodes: 1, size left: 0
Exception in thread "main" java.lang.NoSuchMethodError: ml.dmlc.xgboost4j.scala.spark.XGBoost$.trainWithDataFrame$default$5()Lml/dmlc/xgboost4j/scala/ObjectiveTrait;
	at ml.dmlc.xgboost4j.scala.example.spark.XGBoostTrainer$.main(XGBoostTrainer.scala:120)
	at ml.dmlc.xgboost4j.scala.example.spark.XGBoostTrainer.main(XGBoostTrainer.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
	at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:845)
	at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)
	at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)
	at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
	at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:920)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:929)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
21/02/03 18:34:39 INFO org.spark_project.jetty.server.AbstractConnector: Stopped Spark@4159e81b{HTTP/1.1,[http/1.1]}{0.0.0.0:0}
Job output is complete
```

I then realized that the sample is based on the code from the deprecated component path, which was deleted by https://github.com/kubeflow/pipelines/pull/5045.

Specifically, the not found method from the above error was used here:  
https://github.com/kubeflow/pipelines/blob/32ce8d8f90bfc8f89a2a3c347ad906f99ba776a8/components/deprecated/dataproc/train/src/XGBoostTrainer.scala#L121

And `trainWithDataFrame` only [exists in XGBoost 0.72](https://xgboost.readthedocs.io/en/release_0.72/jvm/scaladocs/xgboost4j-spark/index.html#ml.dmlc.xgboost4j.scala.spark.XGBoost$@trainWithDataFrame(trainingData:org.apache.spark.sql.Dataset[_],params:Map[String,Any],round:Int,nWorkers:Int,obj:ml.dmlc.xgboost4j.scala.ObjectiveTrait,eval:ml.dmlc.xgboost4j.scala.EvalTrait,useExternalMemory:Boolean,missing:Float,featureCol:String,labelCol:String):ml.dmlc.xgboost4j.scala.spark.XGBoostModel), but not seen from any versions beyond. 

XGBoost 0.72 is too old and not even available from https://repo1.maven.org/maven2/com/nvidia/, which is [used by rapids to download XGBoost](https://github.com/GoogleCloudDataproc/initialization-actions/blob/8980d37d16ae580ad1d3eba7a40da59da52ff175/rapids/rapids.sh#L90).

At this point, I feel like we'd rather invest to rewrite the XGBoost sample using the latest XGBoost library than patching the existing one if we do think it's worth demoing running a XGBoost-on-Dataproc pipeline. 
Util we have the sample working, I propose we remove it from the KFP preloaded pipelines and sample-tests.

--- 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/train_until_good/train_until_good.py`
Content:
```
1 #!/usr/bin/env python3
2 # Copyright 2020 The Kubeflow Pipleines authors
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 # This sample demonstrates continuous training using a train-eval-check recursive loop.
17 # The main pipeline trains the initial model and then gradually trains the model
18 # some more until the model evaluation metrics are good enough.
19 
20 import kfp
21 from kfp import components
22 
23 
24 chicago_taxi_dataset_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/e3337b8bdcd63636934954e592d4b32c95b49129/components/datasets/Chicago%20Taxi/component.yaml')
25 xgboost_train_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Train/component.yaml')
26 xgboost_predict_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Predict/component.yaml')
27 
28 pandas_transform_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/6162d55998b176b50267d351241100bb0ee715bc/components/pandas/Transform_DataFrame/in_CSV_format/component.yaml')
29 drop_header_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/02c9638287468c849632cf9f7885b51de4c66f86/components/tables/Remove_header/component.yaml')
30 calculate_regression_metrics_from_csv_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/616542ac0f789914f4eb53438da713dd3004fba4/components/ml_metrics/Calculate_regression_metrics/from_CSV/component.yaml')
31 
32 
33 # This recursive sub-pipeline trains a model, evaluates it, calculates the metrics and checks them.
34 # If the model error is too high, then more training is performed until the model is good.
35 @kfp.dsl.graph_component
36 def train_until_low_error(starting_model, training_data, true_values):
37     # Training
38     model = xgboost_train_on_csv_op(
39         training_data=training_data,
40         starting_model=starting_model,
41         label_column=0,
42         objective='reg:squarederror',
43         num_iterations=50,
44     ).outputs['model']
45 
46     # Predicting
47     predictions = xgboost_predict_on_csv_op(
48         data=training_data,
49         model=model,
50         label_column=0,
51     ).output
52 
53     # Calculating the regression metrics    
54     metrics_task = calculate_regression_metrics_from_csv_op(
55         true_values=true_values,
56         predicted_values=predictions,
57     )
58 
59     # Checking the metrics
60     with kfp.dsl.Condition(metrics_task.outputs['mean_squared_error'] > 0.01):
61         # Training some more
62         train_until_low_error(
63             starting_model=model,
64             training_data=training_data,
65             true_values=true_values,
66         )
67 
68 
69 # The main pipleine trains the initial model and then gradually trains the model some more until the model evaluation metrics are good enough.
70 def train_until_good_pipeline():
71     # Preparing the training data
72     training_data = chicago_taxi_dataset_op(
73         where='trip_start_timestamp >= "2019-01-01" AND trip_start_timestamp < "2019-02-01"',
74         select='tips,trip_seconds,trip_miles,pickup_community_area,dropoff_community_area,fare,tolls,extras,trip_total',
75         limit=10000,
76     ).output
77 
78     # Preparing the true values
79     true_values_table = pandas_transform_csv_op(
80         table=training_data,
81         transform_code='df = df[["tips"]]',
82     ).output
83     
84     true_values = drop_header_op(true_values_table).output
85 
86     # Initial model training
87     first_model = xgboost_train_on_csv_op(
88         training_data=training_data,
89         label_column=0,
90         objective='reg:squarederror',
91         num_iterations=100,
92     ).outputs['model']
93 
94     # Recursively training until the error becomes low
95     train_until_low_error(
96         starting_model=first_model,
97         training_data=training_data,
98         true_values=true_values,
99     )
100 
101 
102 if __name__ == '__main__':
103     kfp_endpoint=None
104     kfp.Client(host=kfp_endpoint).create_run_from_pipeline_func(train_until_good_pipeline, arguments={})
105
```
--- 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/train_until_good/train_until_good.py b/samples/core/train_until_good/train_until_good.py
--- a/samples/core/train_until_good/train_until_good.py
+++ b/samples/core/train_until_good/train_until_good.py
@@ -67,6 +67,7 @@
 
 
 # The main pipleine trains the initial model and then gradually trains the model some more until the model evaluation metrics are good enough.
+@kfp.dsl.pipeline()
 def train_until_good_pipeline():
     # Preparing the training data
     training_data = chicago_taxi_dataset_op(

{"golden_diff": "diff --git a/samples/core/train_until_good/train_until_good.py b/samples/core/train_until_good/train_until_good.py\n--- a/samples/core/train_until_good/train_until_good.py\n+++ b/samples/core/train_until_good/train_until_good.py\n@@ -67,6 +67,7 @@\n \n \n # The main pipleine trains the initial model and then gradually trains the model some more until the model evaluation metrics are good enough.\n+@kfp.dsl.pipeline()\n def train_until_good_pipeline():\n     # Preparing the training data\n     training_data = chicago_taxi_dataset_op(\n", "issue": "The KFP preloaded XGboost sample is broken and out-dated. \n**TL;DR: The preload XGBoost sample is currently broken. \r\nProposing we remove this sample from KFP preload and from sample test until we got a chance to refresh the sample.**\r\n\r\n------------\r\nThe direct cause was that it used the Dataproc 1.2 image which is based on Python 2.7, and pip 21.0 dropped support for Python 2.7.\r\nThe symptom is that `dataproc_create_cluster` fails on initialization. \r\n![image](https://user-images.githubusercontent.com/2043310/106823995-bbcd6780-6636-11eb-8169-92d7a338e048.png)\r\nand the specific error is mentioned [here](https://github.com/kubeflow/pipelines/issues/5007#issuecomment-769637030).\r\n\r\nhttps://github.com/kubeflow/pipelines/pull/5062 made an attempted fix by upgrading to Dataproc 1.5 image. It fixed the Dataproc cluster creation issue, but we hit [an error](https://github.com/kubeflow/pipelines/issues/5007#issuecomment-770182313) later at the Trainer step.\r\n\r\nWe were advised that newer versions of Dataproc images likely don't have XGBoost library preinstalled, as there's now [an initialization action that goes through extra steps to install XGBoost libraries](https://github.com/GoogleCloudDataproc/initialization-actions/tree/master/rapids).\r\n\r\nFollowing that route, I tried installing the default XGBoost version using the rapids script, then hit the error as follows:\r\n```\r\n21/02/03 18:34:20 INFO org.spark_project.jetty.util.log: Logging initialized @3037ms\r\n21/02/03 18:34:20 INFO org.spark_project.jetty.server.Server: jetty-9.3.z-SNAPSHOT, build timestamp: unknown, git hash: unknown\r\n21/02/03 18:34:20 INFO org.spark_project.jetty.server.Server: Started @3169ms\r\n21/02/03 18:34:20 INFO org.spark_project.jetty.server.AbstractConnector: Started ServerConnector@4159e81b{HTTP/1.1,[http/1.1]}{0.0.0.0:37489}\r\n21/02/03 18:34:20 INFO org.apache.hadoop.yarn.client.RMProxy: Connecting to ResourceManager at xgb-bdd8f29b-fb13-4ec2-abcf-38b3699e7ca3-m/10.128.0.101:8032\r\n21/02/03 18:34:21 INFO org.apache.hadoop.yarn.client.AHSProxy: Connecting to Application History server at xgb-bdd8f29b-fb13-4ec2-abcf-38b3699e7ca3-m/10.128.0.101:10200\r\n21/02/03 18:34:21 INFO org.apache.hadoop.conf.Configuration: resource-types.xml not found\r\n21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Unable to find 'resource-types.xml'.\r\n21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Adding resource type - name = memory-mb, units = Mi, type = COUNTABLE\r\n21/02/03 18:34:21 INFO org.apache.hadoop.yarn.util.resource.ResourceUtils: Adding resource type - name = vcores, units = , type = COUNTABLE\r\n21/02/03 18:34:23 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl: Submitted application application_1612377093662_0003\r\n21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1\r\n21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1\r\n21/02/03 18:34:30 INFO org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat: DEBUG: Terminated node allocation with : CompletedNodes: 1, size left: 0\r\n21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1\r\n21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.FileInputFormat: Total input files to process : 1\r\n21/02/03 18:34:36 INFO org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat: DEBUG: Terminated node allocation with : CompletedNodes: 1, size left: 0\r\nException in thread \"main\" java.lang.NoSuchMethodError: ml.dmlc.xgboost4j.scala.spark.XGBoost$.trainWithDataFrame$default$5()Lml/dmlc/xgboost4j/scala/ObjectiveTrait;\r\n\tat ml.dmlc.xgboost4j.scala.example.spark.XGBoostTrainer$.main(XGBoostTrainer.scala:120)\r\n\tat ml.dmlc.xgboost4j.scala.example.spark.XGBoostTrainer.main(XGBoostTrainer.scala)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\tat java.lang.reflect.Method.invoke(Method.java:498)\r\n\tat org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)\r\n\tat org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:845)\r\n\tat org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)\r\n\tat org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)\r\n\tat org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)\r\n\tat org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:920)\r\n\tat org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:929)\r\n\tat org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)\r\n21/02/03 18:34:39 INFO org.spark_project.jetty.server.AbstractConnector: Stopped Spark@4159e81b{HTTP/1.1,[http/1.1]}{0.0.0.0:0}\r\nJob output is complete\r\n```\r\n\r\nI then realized that the sample is based on the code from the deprecated component path, which was deleted by https://github.com/kubeflow/pipelines/pull/5045.\r\n\r\nSpecifically, the not found method from the above error was used here:  \r\nhttps://github.com/kubeflow/pipelines/blob/32ce8d8f90bfc8f89a2a3c347ad906f99ba776a8/components/deprecated/dataproc/train/src/XGBoostTrainer.scala#L121\r\n\r\nAnd `trainWithDataFrame` only [exists in XGBoost 0.72](https://xgboost.readthedocs.io/en/release_0.72/jvm/scaladocs/xgboost4j-spark/index.html#ml.dmlc.xgboost4j.scala.spark.XGBoost$@trainWithDataFrame(trainingData:org.apache.spark.sql.Dataset[_],params:Map[String,Any],round:Int,nWorkers:Int,obj:ml.dmlc.xgboost4j.scala.ObjectiveTrait,eval:ml.dmlc.xgboost4j.scala.EvalTrait,useExternalMemory:Boolean,missing:Float,featureCol:String,labelCol:String):ml.dmlc.xgboost4j.scala.spark.XGBoostModel), but not seen from any versions beyond. \r\n\r\nXGBoost 0.72 is too old and not even available from https://repo1.maven.org/maven2/com/nvidia/, which is [used by rapids to download XGBoost](https://github.com/GoogleCloudDataproc/initialization-actions/blob/8980d37d16ae580ad1d3eba7a40da59da52ff175/rapids/rapids.sh#L90).\r\n\r\nAt this point, I feel like we'd rather invest to rewrite the XGBoost sample using the latest XGBoost library than patching the existing one if we do think it's worth demoing running a XGBoost-on-Dataproc pipeline. \r\nUtil we have the sample working, I propose we remove it from the KFP preloaded pipelines and sample-tests.\n", "before_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2020 The Kubeflow Pipleines authors\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\n# This sample demonstrates continuous training using a train-eval-check recursive loop.\n# The main pipeline trains the initial model and then gradually trains the model\n# some more until the model evaluation metrics are good enough.\n\nimport kfp\nfrom kfp import components\n\n\nchicago_taxi_dataset_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/e3337b8bdcd63636934954e592d4b32c95b49129/components/datasets/Chicago%20Taxi/component.yaml')\nxgboost_train_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Train/component.yaml')\nxgboost_predict_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Predict/component.yaml')\n\npandas_transform_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/6162d55998b176b50267d351241100bb0ee715bc/components/pandas/Transform_DataFrame/in_CSV_format/component.yaml')\ndrop_header_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/02c9638287468c849632cf9f7885b51de4c66f86/components/tables/Remove_header/component.yaml')\ncalculate_regression_metrics_from_csv_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/616542ac0f789914f4eb53438da713dd3004fba4/components/ml_metrics/Calculate_regression_metrics/from_CSV/component.yaml')\n\n\n# This recursive sub-pipeline trains a model, evaluates it, calculates the metrics and checks them.\n# If the model error is too high, then more training is performed until the model is good.\n@kfp.dsl.graph_component\ndef train_until_low_error(starting_model, training_data, true_values):\n    # Training\n    model = xgboost_train_on_csv_op(\n        training_data=training_data,\n        starting_model=starting_model,\n        label_column=0,\n        objective='reg:squarederror',\n        num_iterations=50,\n    ).outputs['model']\n\n    # Predicting\n    predictions = xgboost_predict_on_csv_op(\n        data=training_data,\n        model=model,\n        label_column=0,\n    ).output\n\n    # Calculating the regression metrics    \n    metrics_task = calculate_regression_metrics_from_csv_op(\n        true_values=true_values,\n        predicted_values=predictions,\n    )\n\n    # Checking the metrics\n    with kfp.dsl.Condition(metrics_task.outputs['mean_squared_error'] > 0.01):\n        # Training some more\n        train_until_low_error(\n            starting_model=model,\n            training_data=training_data,\n            true_values=true_values,\n        )\n\n\n# The main pipleine trains the initial model and then gradually trains the model some more until the model evaluation metrics are good enough.\ndef train_until_good_pipeline():\n    # Preparing the training data\n    training_data = chicago_taxi_dataset_op(\n        where='trip_start_timestamp >= \"2019-01-01\" AND trip_start_timestamp < \"2019-02-01\"',\n        select='tips,trip_seconds,trip_miles,pickup_community_area,dropoff_community_area,fare,tolls,extras,trip_total',\n        limit=10000,\n    ).output\n\n    # Preparing the true values\n    true_values_table = pandas_transform_csv_op(\n        table=training_data,\n        transform_code='df = df[[\"tips\"]]',\n    ).output\n    \n    true_values = drop_header_op(true_values_table).output\n\n    # Initial model training\n    first_model = xgboost_train_on_csv_op(\n        training_data=training_data,\n        label_column=0,\n        objective='reg:squarederror',\n        num_iterations=100,\n    ).outputs['model']\n\n    # Recursively training until the error becomes low\n    train_until_low_error(\n        starting_model=first_model,\n        training_data=training_data,\n        true_values=true_values,\n    )\n\n\nif __name__ == '__main__':\n    kfp_endpoint=None\n    kfp.Client(host=kfp_endpoint).create_run_from_pipeline_func(train_until_good_pipeline, arguments={})\n", "path": "samples/core/train_until_good/train_until_good.py"}], "after_files": [{"content": "#!/usr/bin/env python3\n# Copyright 2020 The Kubeflow Pipleines authors\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\n# This sample demonstrates continuous training using a train-eval-check recursive loop.\n# The main pipeline trains the initial model and then gradually trains the model\n# some more until the model evaluation metrics are good enough.\n\nimport kfp\nfrom kfp import components\n\n\nchicago_taxi_dataset_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/e3337b8bdcd63636934954e592d4b32c95b49129/components/datasets/Chicago%20Taxi/component.yaml')\nxgboost_train_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Train/component.yaml')\nxgboost_predict_on_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/567c04c51ff00a1ee525b3458425b17adbe3df61/components/XGBoost/Predict/component.yaml')\n\npandas_transform_csv_op = components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/6162d55998b176b50267d351241100bb0ee715bc/components/pandas/Transform_DataFrame/in_CSV_format/component.yaml')\ndrop_header_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/02c9638287468c849632cf9f7885b51de4c66f86/components/tables/Remove_header/component.yaml')\ncalculate_regression_metrics_from_csv_op = kfp.components.load_component_from_url('https://raw.githubusercontent.com/kubeflow/pipelines/616542ac0f789914f4eb53438da713dd3004fba4/components/ml_metrics/Calculate_regression_metrics/from_CSV/component.yaml')\n\n\n# This recursive sub-pipeline trains a model, evaluates it, calculates the metrics and checks them.\n# If the model error is too high, then more training is performed until the model is good.\n@kfp.dsl.graph_component\ndef train_until_low_error(starting_model, training_data, true_values):\n    # Training\n    model = xgboost_train_on_csv_op(\n        training_data=training_data,\n        starting_model=starting_model,\n        label_column=0,\n        objective='reg:squarederror',\n        num_iterations=50,\n    ).outputs['model']\n\n    # Predicting\n    predictions = xgboost_predict_on_csv_op(\n        data=training_data,\n        model=model,\n        label_column=0,\n    ).output\n\n    # Calculating the regression metrics    \n    metrics_task = calculate_regression_metrics_from_csv_op(\n        true_values=true_values,\n        predicted_values=predictions,\n    )\n\n    # Checking the metrics\n    with kfp.dsl.Condition(metrics_task.outputs['mean_squared_error'] > 0.01):\n        # Training some more\n        train_until_low_error(\n            starting_model=model,\n            training_data=training_data,\n            true_values=true_values,\n        )\n\n\n# The main pipleine trains the initial model and then gradually trains the model some more until the model evaluation metrics are good enough.\n@kfp.dsl.pipeline()\ndef train_until_good_pipeline():\n    # Preparing the training data\n    training_data = chicago_taxi_dataset_op(\n        where='trip_start_timestamp >= \"2019-01-01\" AND trip_start_timestamp < \"2019-02-01\"',\n        select='tips,trip_seconds,trip_miles,pickup_community_area,dropoff_community_area,fare,tolls,extras,trip_total',\n        limit=10000,\n    ).output\n\n    # Preparing the true values\n    true_values_table = pandas_transform_csv_op(\n        table=training_data,\n        transform_code='df = df[[\"tips\"]]',\n    ).output\n    \n    true_values = drop_header_op(true_values_table).output\n\n    # Initial model training\n    first_model = xgboost_train_on_csv_op(\n        training_data=training_data,\n        label_column=0,\n        objective='reg:squarederror',\n        num_iterations=100,\n    ).outputs['model']\n\n    # Recursively training until the error becomes low\n    train_until_low_error(\n        starting_model=first_model,\n        training_data=training_data,\n        true_values=true_values,\n    )\n\n\nif __name__ == '__main__':\n    kfp_endpoint=None\n    kfp.Client(host=kfp_endpoint).create_run_from_pipeline_func(train_until_good_pipeline, arguments={})\n", "path": "samples/core/train_until_good/train_until_good.py"}]}