Joe Schmid09/12/2019, 6:32 PM
I'm getting the warning
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 thought the
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.
line would take care of that but clearly not. What's the right way to address that?
Chris White09/12/2019, 7:13 PM
I’d bet you’ll see a
Constant task above
, and that is the task being referenced in the warning — what was your intent with running
? 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
Joe Schmid09/12/2019, 8:02 PM
new_config != config
Chris White09/12/2019, 8:08 PM
line and the Flow should do what you expect
Joe Schmid09/12/2019, 8:22 PM
Chris White09/12/2019, 8:22 PM
Joe Schmid09/12/2019, 8:28 PM
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!!!"
Jeremiah09/12/2019, 9:01 PM
module filters (which might be overkill). I think we originally anticipated
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.
Joe Schmid09/12/2019, 10:50 PM