Alex Furrier
07/20/2021, 10:06 PMflow_a = StartFlowRun(flow_name="flow a", project_name="my project", wait=True)
flow_b = StartFlowRun(flow_name="flow b", project_name="my project", wait=True)
with Flow('flow-of-flows') as flow_of_flows:
param_a = Parameter('param_a', 'a')
param_b = Parameter('param_b', 'b')
flow_a(parameters={"param_a": param_a})
flow_b(parameters={"param_b": param_b}, upstream_tasks=[flow_a])
In the UI if I attempt to run this flow with params like {'param_a': 'parametrized_1', 'param_b':'parametrized_b'}
it will create a task running flow_a
and fail. Viewing the initiated flow for flow_a
it shows the parameters being {'param_a': null}
. flow_a
fails and flow_b
fails before it even runs since it's dependent on flow_a
and says a reference task failed.
Is there a proper way to pass parameters between parent and child flows?Kevin Kho
07/20/2021, 10:10 PMflow a
in the flow block?Alex Furrier
07/20/2021, 10:17 PM@task
def task_a(a):
return a
with Flow("flow a") as flow:
param_a = Parameter("param_a", "a")
a = task_a(param_a)
Kevin Kho
07/20/2021, 10:20 PMStartFlowRun(flow_name="flow a", project_name="my project", wait=True).run(parameters={"param_a": "param_a_value"})
without the Flow and see if the parameters gets injected? Are you running these with the default or setting them through the UI?Alex Furrier
07/20/2021, 10:47 PMwith Flow('flow-of-flows') as flow_of_flows:
param_a = Parameter('param_a', 'a')
param_b = Parameter('param_b', 'b')
flow_a(parameters={"param_a": param_a})
flow_b(parameters={"param_b": param_b}, upstream_tasks=[flow_a])
it should have been
with Flow('flow-of-flows') as flow_of_flows:
param_a = Parameter('param_a', 'a')
param_b = Parameter('param_b', 'b')
result_a = flow_a(parameters={"param_a": param_a})
flow_b(parameters={"param_b": param_b}, upstream_tasks=[result_a])
Kevin Kho
07/20/2021, 10:49 PM