https://prefect.io
Join Slack
Hello everybody. I have a Flow that create_flow_run (a few) and waits until the finish - wait_for_fl...
# ask-community
m
Hello everybody. I have a Flow that create_flow_run (a few) and waits until the finish - wait_for_flow_run. But if one of the sub-flow is Failed parent flow has a Success state anyway. Can I make parent flow Failed if one of the children fails?
k
Hey @Marko Herkaliuk, I see, maybe you can try the 
StartFlowRun
task with 
StartFlowRun(…, wait=True)
? I think this should fail the flow as it returns the state.
Hey, just chatted with the team and the task state should match the child flow run state. But it will only affect the parent for if it’s a reference task so you need to set it as a reference task if you want it to fail the flow
m
Thanks, Kevin!
n
Hi @Kevin Kho I am having the same problem as Mark. I tried setting the reference tasks in the parent flow of my fof with "parent_flow.set_reference_tasks([hello_run_id, world_run_id]) but even when the both child flows fail, the parent flow succeeds. Is there something I am missing?
k
Did you use 
create_flow_run
or 
StartFlowRun
?
n
Yes I used create_flow_run but not StartFlowRun
k
Ah I see. I think 
StartFlowRun
might be better for you with 
wait=True
. Could you try that? I think it propagates the state
Code for that
n
StartFlowRun is working in terms of getting the parent flow to fail, but now my child flows are running all at once. How can I get them to run in order? @Kevin Kho
k
Are you using a LocalDaskExecutor?
n
I'm just using the default local executor I believe
k
That should be sequential. Maybe you can just set upstream dependencies in the order which you want them to run?
n
that's the .set_upstream method right? where's the api for that?
k
You can do
Copy code
with Flow(...) as flow:
     a = StartFlowRun(...)()
     b = StartFlowRun(...)(upstream_tasks=[a])
or
Copy code
with Flow(...) as flow:
     a = StartFlowRun(...)()
     b = StartFlowRun(...)()

     b.set_upstream(a)
n
perfect thank you @Kevin Kho for all your help !
👍 1
k
Hey @Kevin Kho - following up on this thread as I believe my question is related... in the docs https://docs.prefect.io/core/idioms/flow-to-flow.html it provides the example of initiating flow_a-->d StartFlowRun tasks. The one thing I'd like to confirm is that by including flow_a as an upstream task to flow_b means that it will start flow_a even though its not listed as a task within the actual Flow?
Copy code
with Flow("parent-flow", schedule=weekday_schedule) as flow:
    b = flow_b(upstream_tasks=[flow_a])
    c = flow_c(upstream_tasks=[flow_a])
    d = flow_d(upstream_tasks=[b, c])
vs,
Copy code
with Flow("parent-flow", schedule=weekday_schedule) as flow:
    flow_a = flow_a()
    b = flow_b(upstream_tasks=[flow_a])
    c = flow_c(upstream_tasks=[flow_a])
    d = flow_d(upstream_tasks=[b, c])
k
I believe it will because this will work for other tasks as well where you can pass them by the name to 
upstream_tasks
and it will take care of adding it to the Flow but I honestly prefer the second because it is explicit
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by