feat(sdk): support local pipeline execution #localexecution

[connor-mccarthy]

Description of your changes:
Enables running a pipeline locally:

from kfp import dsl
from kfp import local

local.init(runner=local.SubprocessRunner())


@dsl.component
def add(a: int, b: int) -> int:
    return a + b


@dsl.pipeline
def my_pipeline(num: int) -> int:
    t1 = add(a=num, b=2)
    t2 = add(a=t1.output, b=3)
    return t2.output


task = my_pipeline(num=1)
print(f'Got result: {task.output}')

Currently:

• Only sequential pipelines are supported. Control flow features (dsl.Condition, dsl.ParallelFor, and dsl.ExitHandler) are not supported yet.
• dsl.importeris not supported yet.
• Pipeline-in-pipeline is not supported yet.
• Passing data via f-strings is not supported.

Checklist:

• The title for your pull request (PR) should follow our title convention. Learn more about the pull request title convention used in this repository.

[google-oss-prow]

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

[connor-mccarthy]

/assign @chensun

[connor-mccarthy]

/retest

[connor-mccarthy]

/retest

[connor-mccarthy]

/retest

[chensun]

/lgtm

[chensun]

The two branches possibly need to be merged, otherwise I suspect we may run into bugs for components with DagSpec.

[connor-mccarthy]

I think this LOC should not be hit for the case of a pipeline in a pipeline, since execute_locally would be False (passed in here). For condition and iteration DagSpecs, there is no associated PipelineTask so this LOC would also not be hit.

Is there another case you had in mind?

In my WIP implementation of pipeline-in-pipeline local execution, I've also not hit an issue in this location.

[chensun]

This interface implies the failed task must be from the the top level? A better option would be using a chain of names in case the task failed is nested in layers of dags.

[connor-mccarthy]

I think that makes sense. Since in the current PR this would only amount to wrapping this single element in a list, can we wait to implement this until the implementation of local pipelines-in-pipelines? I think that specific feature will make the motivation for such a data structure clear.

[chensun]

/lgtm
/approve

Thanks!

[google-oss-prow]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: chensun

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:

• sdk/OWNERS [chensun]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[google-oss-prow]

@connor-mccarthy: The following tests failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
kubeflow-pipelines-sdk-execution-tests	c0be34c	link	true	/test kubeflow-pipelines-sdk-execution-tests
kfp-kubernetes-execution-tests	c0be34c	link	false	/test kfp-kubernetes-execution-tests

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. I understand the commands that are listed here.