Hello! I’m trying to setup dependent flows on ECS with Dask following the example at https://docs-v1.prefect.io/core/idioms/flow-to-flow.html I expected my children flows to each run on a dedicated dask worker but they run on the same machine as the parent flow. (code in thread) Anyone can point me to the proper approach?

this Dask executor would need to be applied to the child flow referenced in the create_flow_run task, currently you attach it to the parent

Thanks @Anna Geller. When I attach it to the child flow it starts as many flow and scheduler Fargate instances.

I understand what you try to do now, but afaik Dask cloud provider doesn't work that way. You can submit tasks to Dask cluster running on Fargate but you can't allocate flow runs to separate worker nodes, it's not as simple because Dask scheduler is the part that makes decision where each function call gets submitted to. You can check Dask docs about that or ask in https://dask.discourse.group

parent.py

with Flow(
        name='parent',
        run_config=ECSRun(
            image='MY_IMAGE',
            labels=['my-label']
        ),
        executor=DaskExecutor(cluster_class=lambda: ECSCluster(**{my_args}), adapt_kwargs={'minimum': 0})
        ) as parent:

    input_list = get_my_inputs()

    sub_flows = create_flow_run.map(project_name=unmapped('project'), flow_name=unmapped('child'), [{'input': input} for input in input_list])

    wait_for_flow_run.map(sub_flows,
                          raise_final_state=unmapped(True),
                          stream_logs=unmapped(True))

child.py

with Flow(name='child') as child:
    
    input = Parameter('input')
    
    result1 = task1(input)
    
    result2 = task2(result1)