https://prefect.io logo
Title
j

Joe Schmid

09/12/2019, 6:32 PM
Somebody set me straight on something silly I must be doing wrong. I have a really simple Flow that looks like this:
with Flow("ScaleDownDaskOffHours", environment=env, schedule=schedule) as flow:
    config = get_dask_scheduler_config()
    new_config = reset_config(config)
    save_results = save_new_config(new_config)
    save_results.set_upstream(None, flow)
    ifelse(new_config != config, save_results, None)
I'm getting the warning
PrefectWarning: One of the tasks passed to the switch condition has upstream dependencies: <Task: save_new_config>. Those upstream tasks could run even if the switch condition fails, which might cause unexpected results.
I thought the
save_results.set_upstream(None, flow)
line would take care of that but clearly not. What's the right way to address that?
c

Chris White

09/12/2019, 7:13 PM
Definitely not silly — If you check out
flow.visualize()
I’d bet you’ll see a
None
Constant task above
save_results
, and that is the task being referenced in the warning — what was your intent with running
save_results.set_upstream(None, flow)
? Prefect interprets that first argument as something that should be promoted to a Prefect Task, and in this case that task is simply something which returns the value
None
j

Joe Schmid

09/12/2019, 8:02 PM
I just want the save_new_config/save_results Task to run only if the condition
new_config != config
is True
c

Chris White

09/12/2019, 8:08 PM
OK cool, then you can remove the
save_results.set_upstream(None, flow)
line and the Flow should do what you expect
(even with that line, it should still run as you expect, just with an additional “Constant” task that isn’t really necessary)
j

Joe Schmid

09/12/2019, 8:22 PM
Makes sense and it does run fine, though how do I avoid that warning? (Not a big deal at all but interested to know. 🙂)
c

Chris White

09/12/2019, 8:22 PM
If you remove that line, I’d expect the warning to go away
j

Joe Schmid

09/12/2019, 8:28 PM
Just tried removing that line and still get the warning when doing flow.deploy() (I added that
set_upstream
line after getting the warning.) The warning does make some sense to me -- it's basically saying "hey, the save_new_config task has reset_config as an upstream task -- that might get you weird results since you're using save_new_config/save_results in a switch/ifelse so make up your mind -- do you want it to run when reset_config completes or when your ifelse() condition is True!!!"
Just tried to make a minimal example and don't get the warning. @Chris White don't do anything else on this since I don't want to waste your time. I'll see if I can whittle down my Flow that produces the warning into a minimal example. Sorry for the possible false alarm!
👍 1
j

Jeremiah

09/12/2019, 9:01 PM
Hey Joe — it’s a good question and I guess we really haven’t added a way to turn off this message other than using
warnings
module filters (which might be overkill). I think we originally anticipated
ifelse
being used to kick off one of two task branches (which is an Airflow-ish pattern), rather than being used to gate progress on a single branch, but as you point out this is clearly a good use case — and an annoying one to be warned about.
j

Joe Schmid

09/12/2019, 10:50 PM
Thanks, Jeremiah. No worries at all. The warning is easy enough to ignore and I'm not seeing it in simpler examples. We're mainly concerned with getting the logic of flows right and especially avoiding unexpected gotchas, like that warning is intended to help identify. Chris's advice on just visualizing the graph of the flow makes perfect sense. We're also trying to be thoughtful about when we use the functional approach to building flows vs. imperative, but we're off to a great start.