feat(sdk): Support loading pipeline from yaml #8209
Conversation
|
/test kubeflow-pipelines-samples-v2 |
|
/test kubeflow-pipelines-samples-v2 |
|
/test kubeflow-pipelines-samples-v2 |
1 similar comment
|
/test kubeflow-pipelines-samples-v2 |
|
Reviewing this 👀 |
There was a problem hiding this comment.
Thanks, @chensun. I think this there may be an idempotency issue or at least something I have a question about.
This implementation results in changing PipelineSpec when you reload the pipeline. Try appending the following to one of the pipeline files, running it, then inspecting the one.yaml and two.yaml files. They are different.
if __name__ == '__main__':
compiler.Compiler().compile(
pipeline_func=my_pipeline,
package_path='one.yaml')
from kfp import components
with open('one.yaml', 'r') as f:
reloaded_pipeline = components.load_component_from_text(f.read())
compiler.Compiler().compile(
reloaded_pipeline, package_path='two.yaml')I don't think this is desired, correct? Even if the underlying pipeline logic is itself correct. WDYT?
If I am correct, the read/write tests will be very helpful for validating correctness across a diversity of pipelines if you switch the .pipelines.config.read flag to True, plus make a few other changes (see this commit for the additional changes needed: connor-mccarthy@746d787). If I am not correct, we probably still want to use the read/write tests to validate correctness, but they will need more attention in order to do so.
That's a good point. Although the changed PipelineSpec is still correct, agree that it is ideal to maintain consistency across read and write. Made the change. |
|
All test passed. Merging this to unblock my next PR. |
|
[APPROVALNOTIFIER] This PR is APPROVED Approval requirements bypassed by manually added approval. This pull-request has been approved by: The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
There was a problem hiding this comment.
This looks great, @chensun. Thank you for updating the tests!
/lgtm
| def from_deployment_spec_dict(cls, deployment_spec_dict: Dict[str, Any], | ||
| component_name: str) -> 'Implementation': | ||
| def from_pipeline_spec_dict(cls, pipeline_spec_dict: Dict[str, Any], | ||
| component_name: str) -> 'Implementation': | ||
| """Creates an Implmentation object from a deployment spec message in |
There was a problem hiding this comment.
I think docstrings are out of date
| else: | ||
| pipeline_spec = json_format.ParseDict( | ||
| pipeline_spec_dict, pipeline_spec_pb2.PipelineSpec()) | ||
| return Implementation(graph=pipeline_spec) |
|
|
||
|
|
||
| class YamlComponent(components.BaseComponent): | ||
| """A component loaded from a YAML file. | ||
|
|
||
| **Note:** ``YamlComponent`` is not intended to be used to construct components directly. Use ``kfp.components.load_component_from_*()`` instead. | ||
|
|
||
| Args: | ||
| Artribute: |
There was a problem hiding this comment.
| Artribute: | |
| Attributes: |
| @@ -511,18 +523,16 @@ def transform_outputs(self) -> None: | |||
| def validate_placeholders(self): | |||
| """Validates that input/output placeholders refer to an existing | |||
| input/output.""" | |||
| implementation = self.implementation | |||
| if getattr(implementation, 'container', None) is None: | |||
| if self.implementation.container is None: | |||
| @@ -1470,12 +1488,11 @@ def _rename_component_refs( | |||
| sub_pipeline_spec.root) | |||
|
|
|||
|
|
|||
| def create_pipeline_spec_and_deployment_config( | |||
| def create_pipeline_spec( | |||
* support loading pipeline from yaml * release note * cleanup * maintain read/write consistency with component/pipeline compilation.
Description of your changes:
Checklist: