Problem
Normally, when you use the create_flow_run
task, your child flow run will be displayed in the UI as a task with name create_flow_run
, which may be especially unhelpful if you have multiple child flows that you trigger from a parent flow. This example illustrates the issue:
from prefect import Flow, task
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
FLOW_NAME = "parent_flow_example"
@task(log_stdout=True)
def hello_world():
print(f"Hello from {FLOW_NAME}!")
return FLOW_NAME
with Flow(FLOW_NAME) as flow:
normal_non_subflow_task = hello_world()
first_child_flow_run_id = create_flow_run(
flow_name="child_flow_name",
project_name="PREFECT_PROJECT_NAME",
upstream_tasks=[normal_non_subflow_task],
)
first_child_flowrunview = wait_for_flow_run(
first_child_flow_run_id, raise_final_state=True, stream_logs=True,
)
if __name__ == "__main__":
flow.visualize()
Solution
You can set a custom name using the task_args=dict(name="your_custom_task_name"),
argument available on any Prefect task.
from prefect import Flow, task
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
FLOW_NAME = "parent_flow_example"
@task(log_stdout=True)
def hello_world():
print(f"Hello from {FLOW_NAME}!")
return FLOW_NAME
with Flow(FLOW_NAME) as flow:
normal_non_subflow_task = hello_world()
first_child_flow_run_id = create_flow_run(
flow_name="child_flow_name",
project_name="PREFECT_PROJECT_NAME",
task_args=dict(name="Name it as you like and this will show in the DAG"),
upstream_tasks=[normal_non_subflow_task],
)
first_child_flowrunview = wait_for_flow_run(
first_child_flow_run_id,
raise_final_state=True,
stream_logs=True,
task_args=dict(name="Wait for the first subflow"),
)
if __name__ == "__main__":
flow.visualize()
Other resources
If you would also like to override the randomly generate flow run name (which by default creates flow run names with a random adjective + animal combination such as “amazing-chicken” ) for the child flow, you can use the run_name
argument on the create_flow_run
task.
There is even a way how you can set this name dynamically using a separate task. This topic shows how to do it: