replacement of components with "@graph_component" decorator in V2

[cs-gennadiy]

Please advise or correct me in following Use Case:

With kfp V1:

1. componentA: submits spark application by container with kubectl apply -f spark_app_manifest.yml to spark operator
2. componentB (task inner to componentC): gets application state by kubectl get SparkApplication myApp ...
3. componentC (with "@graph_component" decorator): creates and runs task from componentB. Checks application state - if it isn't equal to "COMPLETED", then runs itself recursively.

With kfp V2:

the "@graph_component" decorator doesn't exist, so there is no way to run a task recursively. A workaround would be to implement "get and check state" loop, like "while" loop, inside of the same component function or command (E.g. using of "dsl.If" wouldn't be possible then).

Is my assumption for V2 correct?

[rimolive]

I never user @graph_component before, but a quick check in SDK code makes me thing that the @graph_component was replaced by @dsl.pipeline, so you can decorate multiple functions with @dsl.pipeline and call them inside the main one.

pipelines/sdk/python/kfp/dsl/graph_component.py

Line 29 in 54f2e45

"""A component defined via @dsl.pipeline decorator.

[cs-gennadiy]

using of @graph_component is in recursion and train_until_good examples.

I tried to model it with following code:

import time
from kfp import dsl

@dsl.component
def component_A(status: dsl.OutputPath(str)):
    # wait for 10 sec
    time.sleep(10)
    # get status
    return "GOOD"

@dsl.pipeline
def pipeline_A():
    # run main pipeline
    pipeline_B()

@dsl.pipeline
def pipeline_B():
    # get status
    status = component_A().outputs["status"]
    # check status
    with dsl.If(status != "GOOD"):
        pipeline_A()

if __name__ == "__main__":
    import kfp.compiler as compiler

    compiler.Compiler().compile(pipeline_A, "pipeline_A.yaml")
    compiler.Compiler().compile(pipeline_B, "pipeline_B.yaml")    

And it isn't get compiled because of cross-dependencies between pipeline_A and pipeline_B.

Generally a component with "@dsl.pipeline" is just a pipeline (a part of graph of main pipeline) which can't run itself recursively. Could you maybe provide a simple example in pseudocode for your idea, please?

[rimolive]

I think it's better if you provide a simple example of your v1 pipeline so we can help you with migrating the code to v2.

[cs-gennadiy]

sure, there is a simplified pipeline v1 (compiled with kfp==1.8.22) with a recursive call of the component with @graph_component decorator:

from kfp import dsl
from kfp.components import create_component_from_func

@create_component_from_func
def get_state() -> str:
    import random
    import time

    # wait for 10 sec
    time.sleep(10)
    # get state
    return random.choice(
        [
            "GOOD",
            "BAD",
            "BAD",
            "BAD",
            "BAD",
            "BAD",
        ]
    )

@dsl.graph_component
def wait_for_state():
    # get state
    get_state_op = get_state()
    get_state_op.execution_options.caching_strategy.max_cache_staleness = "P0D"

    state = get_state_op.outputs["Output"]
    # check state
    with dsl.Condition(state != "GOOD"):
        wait_for_state()

@dsl.pipeline(name="test_v1")
def pipeline_func():
    wait_for_state()

if __name__ == "__main__":
    import kfp.compiler as compiler

    compiler.Compiler().compile(pipeline_func, "pipeline_v1.yaml")

The question is - whether it's possible to migrate such kind of pipeline to v2 and keep the condition check (dsl.Condition()) on pipeline level.

I only succeed it by moving "get state, wait and check state" from pipeline level into get_state() function.

[cs-gennadiy]

hey @rimolive , short question - should I maybe open an issue on this subject?

[rimolive]

Wait a bit. Let me work on a sample to check its funcionality, also if you can join today's Pipelines WG meeting that would be great!

[rimolive]

@cs-gennadiy Speaking to the original author of that decorator, he gave me the full history:
Initially, they thought the @graph_component for recursive task runs would be a good idea, but it didn't get much traction in kfpv1. They added the class as a super type of @dsl.pipeline so people can use it to define sub-DAGs inside their pipelines code, but the recursive feature wasn't implemented because they thought people wouldn't care.

If you think this is a misunderstanding and recursive task runs should be a must have feature, we can convert this discussion in a feature request issue.

[cs-gennadiy]

@rimolive thanks a lot for updating, I was just too late to the meeting.
If there is not much efforts, I would say, yes, it could be a very useful feature for use cases, like I've listed one at the begin of our discussion with checking the state of a Spark Job.

The generalisation of this use case:

• one task starts an asynchronous process.
• downstream task checks for state of this process in loop and waits for the desired state (like "while" in programming language)

Another more concrete MLOps use case would be "train until good" example.

Other benefit of having it - simplified migration from v1 to v2 code (BTW this change isn't mentioned under breaking changes).

Also worth to mention, that KFP v1 is the only orchestration framework which does support a recursion (AFAIK). It isn't possible to implement in Airflow, Luigi or Dagster (again AFAIK), because they are strict on supporting DAG-workflow only.

It just my 2 Cents, maybe there are some other opinions as well.

[rimolive]

Issue #10748 created. We can move the discussion there.

[rimolive]

At last, please mark this discussion as answered.