Hi everyone can someone enlighten this behaviour to me I hav Prefect Community #ask-community

Hi everyone, can someone enlighten this behaviour ...

FuETL

01/11/2023, 6:20 PM

Hi everyone, can someone enlighten this behaviour to me? I have the

case

with flow when is

true

and when is

false

But i noticed that when the case is

false

the

task_b()

is always SKIPPED. After some struggling 🥹 i figure out that the reason that the task was being SKIPPED because inside the case

true

i'm changing the

task_result

variable (if i rename the task run on false cond). Is this the desired behaviour? Why is like that? Is there a way to bypass without renaming?

Copy code

from prefect import task, Flow, case


@task()
def task_a():
    print("task_a")
    return "task_a"


@task()
def task_condition():
    return False


@task()
def task_b(value: str):
    print("task_b")
    return f"task_b={value}"


@task()
def task_c():
    print("task_c")


with Flow("test-flow") as flow:
    task_result = task_a()
    cond = task_condition()

    with case(cond, True):
        task_result = task_b(task_result)
        task_c()

    with case(cond, False):
        task_result = task_b(task_result)

flow.run()

FuETL

01/12/2023, 12:35 AM

prefect duck Sorry to bump, appreciate any help with this. Thank you!

Zanie

01/12/2023, 12:41 AM

Hi.

Zanie

01/12/2023, 12:41 AM

with case: …

always executes when constructing your DAG

Zanie

01/12/2023, 12:42 AM

Your reassignment at

task_result = task_b(task_result)

will always override that variable to be the possibly skipped result of

task_b

Zanie

01/12/2023, 12:42 AM

I would strongly recommend not reusing variable names like this as it will lead to confusing behavior and make it hard to understand what’s going on in your flow.

FuETL

01/12/2023, 1:16 AM

Thanks for the response @Zanie . So in my flow the usage is like: I have a condition that depending of what evaluate i will change the content of the variable in this case

task_result

, so this happen because of the way that DAG is contructed right? I will probably need to read further more about how DAG are constructed (i appreciate if you have some material to share). For sure i will avoid this type of shenanigans in future. 😅

FuETL

01/12/2023, 1:21 AM

DAG with the current code:

FuETL

01/12/2023, 1:22 AM

DAG with different variable names:

Zanie

01/12/2023, 3:23 AM

You can’t change the DAG like that with

case

statements — they’re evaluated at runtime not when you are declaring the flow.

Zanie

01/12/2023, 3:25 AM

Or rather, while declaring the flow both cases are executed to construct the DAG. You can’t have the DAG come back together afterwards based on the value without doing something a little more complicated.

Zanie

01/12/2023, 3:29 AM

You might be able to do something like this

Zanie

01/12/2023, 3:29 AM

Copy code

from prefect import task, Flow, case
from prefect.triggers import not_all_skipped

@task()
def task_a():
    print("task_a")
    return "task_a"


@task()
def task_condition():
    return False


@task()
def task_b(value: str):
    print("task_b")
    return f"task_b={value}"


@task(trigger=not_all_skipped)
def get_case(a, b):
    # select one that's not skipped (not implemented)
    return a


@task
def task_c(value):
    print(value)


with Flow("test-flow") as flow:
    a = task_a()
    cond = task_condition()

    with case(cond, True):
        b_true = task_b(a)

    with case(cond, False):
        b_false = task_b(a)

    b = get_case(b_true, b_false)
    task_c(b)

FuETL

01/12/2023, 11:15 AM

Very nice! Thank you @Zanie

FuETL

01/12/2023, 11:26 AM

Also is possible to use merge as well like:

Copy code

with case(cond, True):
        result1 = task_b("_result1;" + result)
        task_c()

    with case(cond, False):
        result2 = task_b("_result2;" + result)

    result = merge(result1, result2)
    another_task(result)

Zanie

01/12/2023, 5:48 PM

Oh yeah

merge

— I figured we had something like this but couldn’t remember what it was called

Zanie

01/12/2023, 5:48 PM

🙂

4 Views

Open in Slack

Previous Next