When DaskExecutor is used does Prefect delegate DAG of tasks

Question

When DaskExecutor is used does Prefect delegate DAG of tasks scheduling all to DaskScheduler

Kevin Kho · Answer

I think the answer here is not quite Tasks are submitted immediately using `client submit ` The exception is mapping where if you have A gt B gt C then A1 gt B1 gt C1 are submitted The Prefect DAG still has features the Dask DAG doesn t have For example Prefect is still responsible for stuff like caching across flow runs or triggers to determine whether or not to run a task

Kevin Kho · Answer

Stuff like this happens even before submission to Dask

Shiyu Gan · Answer

Consider a simple DAG A gt B Who Prefect or Dask takes care of the dependency scheduling that ensure B only starts after A is done

Kevin Kho · Answer

Prefect

Kevin Kho · Answer

Ah well for mapping like the example above it s Dask But if not mapping it s Prefect

Shiyu Gan · Answer

gt ```Tasks are submitted immediately using client submit ``` If tasks are submitted immediately how is the dependency wait honoured

Kevin Kho · Answer

Because Prefect knows the DAG so it can traverse that DAG during run time and then submit the next task

Shiyu Gan · Answer

So prefect blocks until A is complete then schedules B

Kevin Kho · Answer

Yes because if A fails or raises SKIP then Prefect doesn t want to execute B

Shiyu Gan · Answer

Which line in `flow runner py` s `get flow run state` is the block logic

Kevin Kho · Answer

It s not that simple because there are many checks and the flow runner is pretty hard to read I can try looking for it but is there something you are trying to do

Shiyu Gan · Answer

Can you give me some pointers I can go do the reading myself Just trying to understand prefect implementation code

Shiyu Gan · Answer

Thanks pray

Kevin Kho · Answer

So you can start in the FlowRunner and then notice the tasks are sorted I believe that s the order of submission and then there is a bunch of logic for mapping so it goes to be submitted

Kevin Kho · Answer

By that point it s already submitted to the DaskExecutor so you can see the DaskExecutor run

Kevin Kho · Answer

And then notice in the TaskRunner <https github com PrefectHQ prefect blob master src prefect engine task runner py L197 L198|here> this is run recursively and checks for the conditions to run tasks and then actually runs task in at the end after all the checks

Kevin Kho · Answer

So the answer to your question is there is a sort of the tasks in the Flow and then the tasks are submission in order The task runner recursively checks for conditions The topological sort can be found <https github com PrefectHQ prefect blob master src prefect core flow py L869 L938|here>

Kevin Kho · Answer

Then to the first question note this all happens with without Dask If using DaskExecutor `executor submit ` = `client submit ` If using LocalExecutor `executor submit = regular function call`

Shiyu Gan · Answer

Can you elaborate on TaskRunner recursively checks for conditions

Kevin Kho · Answer

Look at all <https github com PrefectHQ prefect blob master src prefect engine task runner py L247 L291|this>

Shiyu Gan · Answer

Ok it seems recursive task run only comes into play if task loops

Tomer Cagan · Answer

Cool thread thanks for that

Shiyu Gan · Answer

< Kevin Kho> Can you task runner only recursively runs a task if the task loops

Kevin Kho · Answer

I don t think so right Because a Failed state that goes to Retry will end up in that same loop I think I d need to dive in more