We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[bug] pipeline parameters in google_cloud_pipeline_components.aiplatform.TabularCreateDatasetOp
### What steps did you take
Create a simple pipeline that tries to create a tabular dataset from a bigquery table.

```
@dsl.pipeline(
    name='train-churn',
    pipeline_root='gs://blah/test1'
)
def pipeline(project_id: str, dataset_id: str, table_id: str):
    query = dedent(f""" 
    <query elided>
    """)
     
    query_done = bigquery_query_op(
        query=query,
        project_id=project_id,
        dataset_id=dataset_id,
        table_id=table_id,
        dataset_location='us'
    )
    
    # todo bq op unfortunately returns nothing. so we compose our own coordinates for dataset creation
    
    bq_source_rendered = f'bq://{str(project_id)}.{str(dataset_id)}.{str(table_id)}'
       
    dataset = gcc.aiplatform.TabularDatasetCreateOp(
        display_name='test_name_1',
        bq_source=bq_source_rendered
    )
    
    dataset.after(query_done)

   <.... rest of pipeline .... >
```

### What happened:
gcp error log:
```
"__main__", mod_spec)
 File "/opt/python3.7/lib/python3.7/runpy.py", line 85, in _run_code
 exec(code, run_globals)
File "/opt/python3.7/lib/python3.7/site-packages/google_cloud_pipeline_components/aiplatform/remote_runner.py", line 284, in <module>
main()
File "/opt/python3.7/lib/python3.7/site-packages/google_cloud_pipeline_components/aiplatform/remote_runner.py", line 268, in main
key, value = arg[2:].split("=")
ValueError: too many values to unpack (expected 2)
```

AI platform runner does not parse and unpack the parameters correctly, fails on dataset creation.

### What did you expect to happen:
Pipeline should complete.

### Environment:
KFP v2 using vertex. (locally compiled pipeline)

*  KFP SDK version: 
```
kfp                              1.6.5
kfp-pipeline-spec                0.1.8
kfp-server-api                   1.6.0
```


### Anything else you would like to add:
<!-- Miscellaneous information that will assist in solving the issue.-->


### Labels
<!-- Please include labels below by uncommenting them to help us better triage issues -->

<!-- /area frontend -->
<!-- /area backend -->
/area sdk
<!-- /area testing -->
<!-- /area samples -->
 /area components 


---

<!-- Don't delete message below to encourage users to support your issue! -->
Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py`
Content:
```
1 # Copyright 2021 The Kubeflow Authors. 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 """Module for remote execution of AI Platform pipeline component."""
15 
16 import argparse
17 import ast
18 from distutils import util as distutil
19 import inspect
20 import json
21 import os
22 from typing import Any, Callable, Dict, Tuple, Type, TypeVar
23 
24 from google.cloud import aiplatform
25 from google_cloud_pipeline_components.aiplatform import utils
26 
27 INIT_KEY = 'init'
28 METHOD_KEY = 'method'
29 AIPLATFORM_API_VERSION = 'v1'
30 
31 RESOURCE_PREFIX = {
32     "bigquery": "bq://",
33     "aiplatform": "aiplatform://",
34     "google_cloud_storage": "gs://",
35     "google_cloud_storage_gcs_fuse": "/gcs/",
36 }
37 
38 
39 def split_args(kwargs: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
40     """Splits args into constructor and method args.
41 
42     Args:
43         kwargs: kwargs with parameter names preprended with init or method
44     Returns:
45         constructor kwargs, method kwargs
46     """
47     init_args = {}
48     method_args = {}
49 
50     for key, arg in kwargs.items():
51         if key.startswith(INIT_KEY):
52             init_args[key.split(".")[-1]] = arg
53         elif key.startswith(METHOD_KEY):
54             method_args[key.split(".")[-1]] = arg
55 
56     return init_args, method_args
57 
58 
59 def write_to_artifact(executor_input, text):
60     """Write output to local artifact and metadata path (uses GCSFuse)."""
61 
62     output_artifacts = {}
63     for name, artifacts in executor_input.get('outputs', {}).get('artifacts',
64                                                                  {}).items():
65         artifacts_list = artifacts.get('artifacts')
66         if artifacts_list:
67             output_artifacts[name] = artifacts_list[0]
68 
69     executor_output = {}
70     if output_artifacts:
71         executor_output['artifacts'] = {}
72         uri_with_prefix = ""
73 
74         # TODO - Support multiple outputs, current implementation.
75 
76         # sets all output uri's to text
77         for name, artifact in output_artifacts.items():
78             # Add URI Prefix
79             # "aiplatform://API_VERSION/": For AI Platform resource names, current version is defined in AIPLATFORM_API_VERSION.
80             if aiplatform.utils.RESOURCE_NAME_PATTERN.match(text):
81                 uri_with_prefix = f"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/{text}"
82 
83             # "gcs://": For Google Cloud Storage resources.
84             elif text.startswith(
85                     RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):
86                 uri_with_prefix = text.replace(
87                     RESOURCE_PREFIX['google_cloud_storage_gcs_fuse'],
88                     RESOURCE_PREFIX.get("google_cloud_storage")
89                 )
90 
91             # "bq://": For BigQuery resources.
92             elif text.startswith(RESOURCE_PREFIX.get('bigquery')):
93                 uri_with_prefix = text
94             else:
95                 uri_with_prefix = text
96 
97             runtime_artifact = {
98                 "name": artifact.get('name'),
99                 "uri": uri_with_prefix,
100                 "metadata": artifact.get('metadata', {})
101             }
102             artifacts_list = {'artifacts': [runtime_artifact]}
103 
104             executor_output['artifacts'][name] = artifacts_list
105 
106     os.makedirs(
107         os.path.dirname(executor_input['outputs']['outputFile']), exist_ok=True
108     )
109     with open(executor_input['outputs']['outputFile'], 'w') as f:
110         f.write(json.dumps(executor_output))
111 
112 
113 def resolve_input_args(value, type_to_resolve):
114     """If this is an input from Pipelines, read it directly from gcs."""
115     if inspect.isclass(type_to_resolve) and issubclass(
116             type_to_resolve, aiplatform.base.VertexAiResourceNoun):
117         # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix
118         if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):
119             value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):
120                          ]
121         # Remove `aiplatform` prefix from resource name
122         if value.startswith(RESOURCE_PREFIX.get("aiplatform")):
123             prefix_str = f"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/"
124             value = value[len(prefix_str):]
125 
126     # No action needed for Google Cloud Storage prefix.
127     # No action needed for BigQuery resource names.
128     return value
129 
130 
131 def resolve_init_args(key, value):
132     """Resolves Metadata/InputPath parameters to resource names."""
133     if key.endswith('_name'):
134         # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix
135         if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):
136             # not a resource noun, remove the /gcs/ prefix
137             value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):
138                          ]
139         # Remove `aiplatform` prefix from resource name
140         if value.startswith(RESOURCE_PREFIX.get("aiplatform")):
141             prefix_str = f"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/"
142             value = value[len(prefix_str):]
143 
144     # No action needed for Google Cloud Storage prefix.
145     # No action needed for BigQuery resource names.
146     return value
147 
148 
149 def make_output(output_object: Any) -> str:
150     if utils.is_mb_sdk_resource_noun_type(type(output_object)):
151         return output_object.resource_name
152 
153     # TODO: handle more default cases
154     # right now this is required for export data because proto Repeated
155     # this should be expanded to handle multiple different types
156     # or possibly export data should return a Dataset
157     return json.dumps(list(output_object))
158 
159 
160 T = TypeVar('T')
161 
162 
163 def cast(value: str, annotation_type: Type[T]) -> T:
164     """Casts a value to the annotation type.
165 
166     Includes special handling for bools passed as strings.
167 
168     Args:
169         value (str): The value represented as a string.
170         annotation_type (Type[T]): The type to cast the value to.
171     Returns:
172         An instance of annotation_type value.
173     """
174     if annotation_type is bool:
175         return bool(distutil.strtobool(value))
176     return annotation_type(value)
177 
178 
179 def prepare_parameters(
180     kwargs: Dict[str, Any], method: Callable, is_init: bool = False
181 ):
182     """Prepares parameters passed into components before calling SDK.
183 
184     1. Determines the annotation type that should used with the parameter
185     2. Reads input values if needed
186     3. Deserializes thos value where appropriate
187     4. Or casts to the correct type.
188 
189     Args:
190         kwargs (Dict[str, Any]): The kwargs that will be passed into method. Mutates in place.
191         method (Callable): The method the kwargs used to invoke the method.
192         is_init (bool): Whether this method is a constructor
193     """
194     for key, param in inspect.signature(method).parameters.items():
195         if key in kwargs:
196             value = kwargs[key]
197             param_type = utils.resolve_annotation(param.annotation)
198             value = resolve_init_args(
199                 key, value
200             ) if is_init else resolve_input_args(value, param_type)
201             deserializer = utils.get_deserializer(param_type)
202             if deserializer:
203                 value = deserializer(value)
204             else:
205                 value = cast(value, param_type)
206 
207             try:
208                 # Attempt at converting String to list:
209                 # Some parameters accept union[str, sequence[str]]
210                 # For these serialization with json is not possible as
211                 # component yaml conversion swaps double and single
212                 # quotes resulting in `JSON.Loads` loading such a list as
213                 # a String. Using ast.literal_eval to attempt to convert the
214                 # str back to a python List.
215                 value = ast.literal_eval(value)
216                 print(f"Conversion for value succeeded for value: {value}")
217             except:
218                 # The input was actually a String and not a List,
219                 # no additional transformations are required.
220                 pass
221 
222             kwargs[key] = value
223 
224 
225 def runner(cls_name, method_name, executor_input, kwargs):
226     cls = getattr(aiplatform, cls_name)
227 
228     init_args, method_args = split_args(kwargs)
229 
230     serialized_args = {INIT_KEY: init_args, METHOD_KEY: method_args}
231 
232     prepare_parameters(serialized_args[INIT_KEY], cls.__init__, is_init=True)
233     obj = cls(**serialized_args[INIT_KEY]) if serialized_args[INIT_KEY] else cls
234 
235     method = getattr(obj, method_name)
236     prepare_parameters(serialized_args[METHOD_KEY], method, is_init=False)
237 
238     print(
239         f"method:{method} is being called with parameters {serialized_args[METHOD_KEY]}"
240     )
241     output = method(**serialized_args[METHOD_KEY])
242 
243     if output:
244         write_to_artifact(executor_input, make_output(output))
245         return output
246 
247 
248 def main():
249     parser = argparse.ArgumentParser()
250     parser.add_argument("--cls_name", type=str)
251     parser.add_argument("--method_name", type=str)
252     parser.add_argument("--executor_input", type=str, default=None)
253 
254     args, unknown_args = parser.parse_known_args()
255     kwargs = {}
256 
257     executor_input = json.loads(args.executor_input)
258 
259     key_value = None
260     for arg in unknown_args:
261 
262         print(arg)
263 
264         # Remove whitespace from arg.
265         arg = arg.strip()
266 
267         if "=" in arg:
268             key, value = arg[2:].split("=")
269             kwargs[key] = value
270         else:
271             if not key_value:
272                 key_value = arg[2:]
273             else:
274                 kwargs[key_value] = arg
275                 key_value = None
276 
277     # Update user agent header for metrics reporting
278     aiplatform.constants.USER_AGENT_PRODUCT = "google-cloud-pipeline-components"
279 
280     print(runner(args.cls_name, args.method_name, executor_input, kwargs))
281 
282 
283 if __name__ == "__main__":
284     main()
285
```
--- 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/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py b/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py
--- a/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py
+++ b/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py
@@ -265,7 +265,7 @@
         arg = arg.strip()
 
         if "=" in arg:
-            key, value = arg[2:].split("=")
+            key, value = arg[2:].split("=", 1)
             kwargs[key] = value
         else:
             if not key_value:

{"golden_diff": "diff --git a/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py b/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py\n--- a/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py\n+++ b/components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py\n@@ -265,7 +265,7 @@\n         arg = arg.strip()\n \n         if \"=\" in arg:\n-            key, value = arg[2:].split(\"=\")\n+            key, value = arg[2:].split(\"=\", 1)\n             kwargs[key] = value\n         else:\n             if not key_value:\n", "issue": "[bug] pipeline parameters in google_cloud_pipeline_components.aiplatform.TabularCreateDatasetOp\n### What steps did you take\r\nCreate a simple pipeline that tries to create a tabular dataset from a bigquery table.\r\n\r\n```\r\n@dsl.pipeline(\r\n    name='train-churn',\r\n    pipeline_root='gs://blah/test1'\r\n)\r\ndef pipeline(project_id: str, dataset_id: str, table_id: str):\r\n    query = dedent(f\"\"\" \r\n    <query elided>\r\n    \"\"\")\r\n     \r\n    query_done = bigquery_query_op(\r\n        query=query,\r\n        project_id=project_id,\r\n        dataset_id=dataset_id,\r\n        table_id=table_id,\r\n        dataset_location='us'\r\n    )\r\n    \r\n    # todo bq op unfortunately returns nothing. so we compose our own coordinates for dataset creation\r\n    \r\n    bq_source_rendered = f'bq://{str(project_id)}.{str(dataset_id)}.{str(table_id)}'\r\n       \r\n    dataset = gcc.aiplatform.TabularDatasetCreateOp(\r\n        display_name='test_name_1',\r\n        bq_source=bq_source_rendered\r\n    )\r\n    \r\n    dataset.after(query_done)\r\n\r\n   <.... rest of pipeline .... >\r\n```\r\n\r\n### What happened:\r\ngcp error log:\r\n```\r\n\"__main__\", mod_spec)\r\n File \"/opt/python3.7/lib/python3.7/runpy.py\", line 85, in _run_code\r\n exec(code, run_globals)\r\nFile \"/opt/python3.7/lib/python3.7/site-packages/google_cloud_pipeline_components/aiplatform/remote_runner.py\", line 284, in <module>\r\nmain()\r\nFile \"/opt/python3.7/lib/python3.7/site-packages/google_cloud_pipeline_components/aiplatform/remote_runner.py\", line 268, in main\r\nkey, value = arg[2:].split(\"=\")\r\nValueError: too many values to unpack (expected 2)\r\n```\r\n\r\nAI platform runner does not parse and unpack the parameters correctly, fails on dataset creation.\r\n\r\n### What did you expect to happen:\r\nPipeline should complete.\r\n\r\n### Environment:\r\nKFP v2 using vertex. (locally compiled pipeline)\r\n\r\n*  KFP SDK version: \r\n```\r\nkfp                              1.6.5\r\nkfp-pipeline-spec                0.1.8\r\nkfp-server-api                   1.6.0\r\n```\r\n\r\n\r\n### Anything else you would like to add:\r\n<!-- Miscellaneous information that will assist in solving the issue.-->\r\n\r\n\r\n### Labels\r\n<!-- Please include labels below by uncommenting them to help us better triage issues -->\r\n\r\n<!-- /area frontend -->\r\n<!-- /area backend -->\r\n/area sdk\r\n<!-- /area testing -->\r\n<!-- /area samples -->\r\n /area components \r\n\r\n\r\n---\r\n\r\n<!-- Don't delete message below to encourage users to support your issue! -->\r\nImpacted by this bug? Give it a \ud83d\udc4d. We prioritise the issues with the most \ud83d\udc4d.\r\n\n", "before_files": [{"content": "# Copyright 2021 The Kubeflow Authors. 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\"\"\"Module for remote execution of AI Platform pipeline component.\"\"\"\n\nimport argparse\nimport ast\nfrom distutils import util as distutil\nimport inspect\nimport json\nimport os\nfrom typing import Any, Callable, Dict, Tuple, Type, TypeVar\n\nfrom google.cloud import aiplatform\nfrom google_cloud_pipeline_components.aiplatform import utils\n\nINIT_KEY = 'init'\nMETHOD_KEY = 'method'\nAIPLATFORM_API_VERSION = 'v1'\n\nRESOURCE_PREFIX = {\n    \"bigquery\": \"bq://\",\n    \"aiplatform\": \"aiplatform://\",\n    \"google_cloud_storage\": \"gs://\",\n    \"google_cloud_storage_gcs_fuse\": \"/gcs/\",\n}\n\n\ndef split_args(kwargs: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:\n    \"\"\"Splits args into constructor and method args.\n\n    Args:\n        kwargs: kwargs with parameter names preprended with init or method\n    Returns:\n        constructor kwargs, method kwargs\n    \"\"\"\n    init_args = {}\n    method_args = {}\n\n    for key, arg in kwargs.items():\n        if key.startswith(INIT_KEY):\n            init_args[key.split(\".\")[-1]] = arg\n        elif key.startswith(METHOD_KEY):\n            method_args[key.split(\".\")[-1]] = arg\n\n    return init_args, method_args\n\n\ndef write_to_artifact(executor_input, text):\n    \"\"\"Write output to local artifact and metadata path (uses GCSFuse).\"\"\"\n\n    output_artifacts = {}\n    for name, artifacts in executor_input.get('outputs', {}).get('artifacts',\n                                                                 {}).items():\n        artifacts_list = artifacts.get('artifacts')\n        if artifacts_list:\n            output_artifacts[name] = artifacts_list[0]\n\n    executor_output = {}\n    if output_artifacts:\n        executor_output['artifacts'] = {}\n        uri_with_prefix = \"\"\n\n        # TODO - Support multiple outputs, current implementation.\n\n        # sets all output uri's to text\n        for name, artifact in output_artifacts.items():\n            # Add URI Prefix\n            # \"aiplatform://API_VERSION/\": For AI Platform resource names, current version is defined in AIPLATFORM_API_VERSION.\n            if aiplatform.utils.RESOURCE_NAME_PATTERN.match(text):\n                uri_with_prefix = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/{text}\"\n\n            # \"gcs://\": For Google Cloud Storage resources.\n            elif text.startswith(\n                    RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                uri_with_prefix = text.replace(\n                    RESOURCE_PREFIX['google_cloud_storage_gcs_fuse'],\n                    RESOURCE_PREFIX.get(\"google_cloud_storage\")\n                )\n\n            # \"bq://\": For BigQuery resources.\n            elif text.startswith(RESOURCE_PREFIX.get('bigquery')):\n                uri_with_prefix = text\n            else:\n                uri_with_prefix = text\n\n            runtime_artifact = {\n                \"name\": artifact.get('name'),\n                \"uri\": uri_with_prefix,\n                \"metadata\": artifact.get('metadata', {})\n            }\n            artifacts_list = {'artifacts': [runtime_artifact]}\n\n            executor_output['artifacts'][name] = artifacts_list\n\n    os.makedirs(\n        os.path.dirname(executor_input['outputs']['outputFile']), exist_ok=True\n    )\n    with open(executor_input['outputs']['outputFile'], 'w') as f:\n        f.write(json.dumps(executor_output))\n\n\ndef resolve_input_args(value, type_to_resolve):\n    \"\"\"If this is an input from Pipelines, read it directly from gcs.\"\"\"\n    if inspect.isclass(type_to_resolve) and issubclass(\n            type_to_resolve, aiplatform.base.VertexAiResourceNoun):\n        # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix\n        if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n            value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                         ]\n        # Remove `aiplatform` prefix from resource name\n        if value.startswith(RESOURCE_PREFIX.get(\"aiplatform\")):\n            prefix_str = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/\"\n            value = value[len(prefix_str):]\n\n    # No action needed for Google Cloud Storage prefix.\n    # No action needed for BigQuery resource names.\n    return value\n\n\ndef resolve_init_args(key, value):\n    \"\"\"Resolves Metadata/InputPath parameters to resource names.\"\"\"\n    if key.endswith('_name'):\n        # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix\n        if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n            # not a resource noun, remove the /gcs/ prefix\n            value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                         ]\n        # Remove `aiplatform` prefix from resource name\n        if value.startswith(RESOURCE_PREFIX.get(\"aiplatform\")):\n            prefix_str = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/\"\n            value = value[len(prefix_str):]\n\n    # No action needed for Google Cloud Storage prefix.\n    # No action needed for BigQuery resource names.\n    return value\n\n\ndef make_output(output_object: Any) -> str:\n    if utils.is_mb_sdk_resource_noun_type(type(output_object)):\n        return output_object.resource_name\n\n    # TODO: handle more default cases\n    # right now this is required for export data because proto Repeated\n    # this should be expanded to handle multiple different types\n    # or possibly export data should return a Dataset\n    return json.dumps(list(output_object))\n\n\nT = TypeVar('T')\n\n\ndef cast(value: str, annotation_type: Type[T]) -> T:\n    \"\"\"Casts a value to the annotation type.\n\n    Includes special handling for bools passed as strings.\n\n    Args:\n        value (str): The value represented as a string.\n        annotation_type (Type[T]): The type to cast the value to.\n    Returns:\n        An instance of annotation_type value.\n    \"\"\"\n    if annotation_type is bool:\n        return bool(distutil.strtobool(value))\n    return annotation_type(value)\n\n\ndef prepare_parameters(\n    kwargs: Dict[str, Any], method: Callable, is_init: bool = False\n):\n    \"\"\"Prepares parameters passed into components before calling SDK.\n\n    1. Determines the annotation type that should used with the parameter\n    2. Reads input values if needed\n    3. Deserializes thos value where appropriate\n    4. Or casts to the correct type.\n\n    Args:\n        kwargs (Dict[str, Any]): The kwargs that will be passed into method. Mutates in place.\n        method (Callable): The method the kwargs used to invoke the method.\n        is_init (bool): Whether this method is a constructor\n    \"\"\"\n    for key, param in inspect.signature(method).parameters.items():\n        if key in kwargs:\n            value = kwargs[key]\n            param_type = utils.resolve_annotation(param.annotation)\n            value = resolve_init_args(\n                key, value\n            ) if is_init else resolve_input_args(value, param_type)\n            deserializer = utils.get_deserializer(param_type)\n            if deserializer:\n                value = deserializer(value)\n            else:\n                value = cast(value, param_type)\n\n            try:\n                # Attempt at converting String to list:\n                # Some parameters accept union[str, sequence[str]]\n                # For these serialization with json is not possible as\n                # component yaml conversion swaps double and single\n                # quotes resulting in `JSON.Loads` loading such a list as\n                # a String. Using ast.literal_eval to attempt to convert the\n                # str back to a python List.\n                value = ast.literal_eval(value)\n                print(f\"Conversion for value succeeded for value: {value}\")\n            except:\n                # The input was actually a String and not a List,\n                # no additional transformations are required.\n                pass\n\n            kwargs[key] = value\n\n\ndef runner(cls_name, method_name, executor_input, kwargs):\n    cls = getattr(aiplatform, cls_name)\n\n    init_args, method_args = split_args(kwargs)\n\n    serialized_args = {INIT_KEY: init_args, METHOD_KEY: method_args}\n\n    prepare_parameters(serialized_args[INIT_KEY], cls.__init__, is_init=True)\n    obj = cls(**serialized_args[INIT_KEY]) if serialized_args[INIT_KEY] else cls\n\n    method = getattr(obj, method_name)\n    prepare_parameters(serialized_args[METHOD_KEY], method, is_init=False)\n\n    print(\n        f\"method:{method} is being called with parameters {serialized_args[METHOD_KEY]}\"\n    )\n    output = method(**serialized_args[METHOD_KEY])\n\n    if output:\n        write_to_artifact(executor_input, make_output(output))\n        return output\n\n\ndef main():\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--cls_name\", type=str)\n    parser.add_argument(\"--method_name\", type=str)\n    parser.add_argument(\"--executor_input\", type=str, default=None)\n\n    args, unknown_args = parser.parse_known_args()\n    kwargs = {}\n\n    executor_input = json.loads(args.executor_input)\n\n    key_value = None\n    for arg in unknown_args:\n\n        print(arg)\n\n        # Remove whitespace from arg.\n        arg = arg.strip()\n\n        if \"=\" in arg:\n            key, value = arg[2:].split(\"=\")\n            kwargs[key] = value\n        else:\n            if not key_value:\n                key_value = arg[2:]\n            else:\n                kwargs[key_value] = arg\n                key_value = None\n\n    # Update user agent header for metrics reporting\n    aiplatform.constants.USER_AGENT_PRODUCT = \"google-cloud-pipeline-components\"\n\n    print(runner(args.cls_name, args.method_name, executor_input, kwargs))\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py"}], "after_files": [{"content": "# Copyright 2021 The Kubeflow Authors. 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\"\"\"Module for remote execution of AI Platform pipeline component.\"\"\"\n\nimport argparse\nimport ast\nfrom distutils import util as distutil\nimport inspect\nimport json\nimport os\nfrom typing import Any, Callable, Dict, Tuple, Type, TypeVar\n\nfrom google.cloud import aiplatform\nfrom google_cloud_pipeline_components.aiplatform import utils\n\nINIT_KEY = 'init'\nMETHOD_KEY = 'method'\nAIPLATFORM_API_VERSION = 'v1'\n\nRESOURCE_PREFIX = {\n    \"bigquery\": \"bq://\",\n    \"aiplatform\": \"aiplatform://\",\n    \"google_cloud_storage\": \"gs://\",\n    \"google_cloud_storage_gcs_fuse\": \"/gcs/\",\n}\n\n\ndef split_args(kwargs: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]:\n    \"\"\"Splits args into constructor and method args.\n\n    Args:\n        kwargs: kwargs with parameter names preprended with init or method\n    Returns:\n        constructor kwargs, method kwargs\n    \"\"\"\n    init_args = {}\n    method_args = {}\n\n    for key, arg in kwargs.items():\n        if key.startswith(INIT_KEY):\n            init_args[key.split(\".\")[-1]] = arg\n        elif key.startswith(METHOD_KEY):\n            method_args[key.split(\".\")[-1]] = arg\n\n    return init_args, method_args\n\n\ndef write_to_artifact(executor_input, text):\n    \"\"\"Write output to local artifact and metadata path (uses GCSFuse).\"\"\"\n\n    output_artifacts = {}\n    for name, artifacts in executor_input.get('outputs', {}).get('artifacts',\n                                                                 {}).items():\n        artifacts_list = artifacts.get('artifacts')\n        if artifacts_list:\n            output_artifacts[name] = artifacts_list[0]\n\n    executor_output = {}\n    if output_artifacts:\n        executor_output['artifacts'] = {}\n        uri_with_prefix = \"\"\n\n        # TODO - Support multiple outputs, current implementation.\n\n        # sets all output uri's to text\n        for name, artifact in output_artifacts.items():\n            # Add URI Prefix\n            # \"aiplatform://API_VERSION/\": For AI Platform resource names, current version is defined in AIPLATFORM_API_VERSION.\n            if aiplatform.utils.RESOURCE_NAME_PATTERN.match(text):\n                uri_with_prefix = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/{text}\"\n\n            # \"gcs://\": For Google Cloud Storage resources.\n            elif text.startswith(\n                    RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                uri_with_prefix = text.replace(\n                    RESOURCE_PREFIX['google_cloud_storage_gcs_fuse'],\n                    RESOURCE_PREFIX.get(\"google_cloud_storage\")\n                )\n\n            # \"bq://\": For BigQuery resources.\n            elif text.startswith(RESOURCE_PREFIX.get('bigquery')):\n                uri_with_prefix = text\n            else:\n                uri_with_prefix = text\n\n            runtime_artifact = {\n                \"name\": artifact.get('name'),\n                \"uri\": uri_with_prefix,\n                \"metadata\": artifact.get('metadata', {})\n            }\n            artifacts_list = {'artifacts': [runtime_artifact]}\n\n            executor_output['artifacts'][name] = artifacts_list\n\n    os.makedirs(\n        os.path.dirname(executor_input['outputs']['outputFile']), exist_ok=True\n    )\n    with open(executor_input['outputs']['outputFile'], 'w') as f:\n        f.write(json.dumps(executor_output))\n\n\ndef resolve_input_args(value, type_to_resolve):\n    \"\"\"If this is an input from Pipelines, read it directly from gcs.\"\"\"\n    if inspect.isclass(type_to_resolve) and issubclass(\n            type_to_resolve, aiplatform.base.VertexAiResourceNoun):\n        # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix\n        if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n            value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                         ]\n        # Remove `aiplatform` prefix from resource name\n        if value.startswith(RESOURCE_PREFIX.get(\"aiplatform\")):\n            prefix_str = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/\"\n            value = value[len(prefix_str):]\n\n    # No action needed for Google Cloud Storage prefix.\n    # No action needed for BigQuery resource names.\n    return value\n\n\ndef resolve_init_args(key, value):\n    \"\"\"Resolves Metadata/InputPath parameters to resource names.\"\"\"\n    if key.endswith('_name'):\n        # Remove '/gcs/' prefix before attempting to remove `aiplatform` prefix\n        if value.startswith(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n            # not a resource noun, remove the /gcs/ prefix\n            value = value[len(RESOURCE_PREFIX['google_cloud_storage_gcs_fuse']):\n                         ]\n        # Remove `aiplatform` prefix from resource name\n        if value.startswith(RESOURCE_PREFIX.get(\"aiplatform\")):\n            prefix_str = f\"{RESOURCE_PREFIX['aiplatform']}{AIPLATFORM_API_VERSION}/\"\n            value = value[len(prefix_str):]\n\n    # No action needed for Google Cloud Storage prefix.\n    # No action needed for BigQuery resource names.\n    return value\n\n\ndef make_output(output_object: Any) -> str:\n    if utils.is_mb_sdk_resource_noun_type(type(output_object)):\n        return output_object.resource_name\n\n    # TODO: handle more default cases\n    # right now this is required for export data because proto Repeated\n    # this should be expanded to handle multiple different types\n    # or possibly export data should return a Dataset\n    return json.dumps(list(output_object))\n\n\nT = TypeVar('T')\n\n\ndef cast(value: str, annotation_type: Type[T]) -> T:\n    \"\"\"Casts a value to the annotation type.\n\n    Includes special handling for bools passed as strings.\n\n    Args:\n        value (str): The value represented as a string.\n        annotation_type (Type[T]): The type to cast the value to.\n    Returns:\n        An instance of annotation_type value.\n    \"\"\"\n    if annotation_type is bool:\n        return bool(distutil.strtobool(value))\n    return annotation_type(value)\n\n\ndef prepare_parameters(\n    kwargs: Dict[str, Any], method: Callable, is_init: bool = False\n):\n    \"\"\"Prepares parameters passed into components before calling SDK.\n\n    1. Determines the annotation type that should used with the parameter\n    2. Reads input values if needed\n    3. Deserializes thos value where appropriate\n    4. Or casts to the correct type.\n\n    Args:\n        kwargs (Dict[str, Any]): The kwargs that will be passed into method. Mutates in place.\n        method (Callable): The method the kwargs used to invoke the method.\n        is_init (bool): Whether this method is a constructor\n    \"\"\"\n    for key, param in inspect.signature(method).parameters.items():\n        if key in kwargs:\n            value = kwargs[key]\n            param_type = utils.resolve_annotation(param.annotation)\n            value = resolve_init_args(\n                key, value\n            ) if is_init else resolve_input_args(value, param_type)\n            deserializer = utils.get_deserializer(param_type)\n            if deserializer:\n                value = deserializer(value)\n            else:\n                value = cast(value, param_type)\n\n            try:\n                # Attempt at converting String to list:\n                # Some parameters accept union[str, sequence[str]]\n                # For these serialization with json is not possible as\n                # component yaml conversion swaps double and single\n                # quotes resulting in `JSON.Loads` loading such a list as\n                # a String. Using ast.literal_eval to attempt to convert the\n                # str back to a python List.\n                value = ast.literal_eval(value)\n                print(f\"Conversion for value succeeded for value: {value}\")\n            except:\n                # The input was actually a String and not a List,\n                # no additional transformations are required.\n                pass\n\n            kwargs[key] = value\n\n\ndef runner(cls_name, method_name, executor_input, kwargs):\n    cls = getattr(aiplatform, cls_name)\n\n    init_args, method_args = split_args(kwargs)\n\n    serialized_args = {INIT_KEY: init_args, METHOD_KEY: method_args}\n\n    prepare_parameters(serialized_args[INIT_KEY], cls.__init__, is_init=True)\n    obj = cls(**serialized_args[INIT_KEY]) if serialized_args[INIT_KEY] else cls\n\n    method = getattr(obj, method_name)\n    prepare_parameters(serialized_args[METHOD_KEY], method, is_init=False)\n\n    print(\n        f\"method:{method} is being called with parameters {serialized_args[METHOD_KEY]}\"\n    )\n    output = method(**serialized_args[METHOD_KEY])\n\n    if output:\n        write_to_artifact(executor_input, make_output(output))\n        return output\n\n\ndef main():\n    parser = argparse.ArgumentParser()\n    parser.add_argument(\"--cls_name\", type=str)\n    parser.add_argument(\"--method_name\", type=str)\n    parser.add_argument(\"--executor_input\", type=str, default=None)\n\n    args, unknown_args = parser.parse_known_args()\n    kwargs = {}\n\n    executor_input = json.loads(args.executor_input)\n\n    key_value = None\n    for arg in unknown_args:\n\n        print(arg)\n\n        # Remove whitespace from arg.\n        arg = arg.strip()\n\n        if \"=\" in arg:\n            key, value = arg[2:].split(\"=\", 1)\n            kwargs[key] = value\n        else:\n            if not key_value:\n                key_value = arg[2:]\n            else:\n                kwargs[key_value] = arg\n                key_value = None\n\n    # Update user agent header for metrics reporting\n    aiplatform.constants.USER_AGENT_PRODUCT = \"google-cloud-pipeline-components\"\n\n    print(runner(args.cls_name, args.method_name, executor_input, kwargs))\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "components/google-cloud/google_cloud_pipeline_components/remote/aiplatform/remote_runner.py"}]}