Hello 🙂 I'm wondering if someone can explain to me how does prefect core execute mapping with the DaskExecuter. In the Docs, it's said that Prefect automatically creates a copy of the task for each element of its input data. However, looking at the code I'm not understanding where that is being done, and when those new tasks are being submitted to Dask.

Hi @Ruben Silva! I think these are two of the critical files:

Task.map

method: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/engine/flow_runner.py

FlowRunner

class: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/engine/flow_runner.py

Hey @Ruben Silva! Fair warning, what you’re looking for is probably some of the more intricate code in the repo and there’s a lot of different moving pieces that come together to make this possible, but this is the entrypoint: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/engine/flow_runner.py#L545

That code it's very different from the version of my fork. It look a lot better, I'll check it out.