Hi I m trying to run my flow from another file importing it

Question

Hi I m trying to run my flow from another file importing it to inject custom state handlers do other stuff ``` gt gt gt from example flow import flow gt gt gt from prefect engine executors import DaskExecutor gt gt gt flow run this works 2019 09 06 20 38 53 240 INFO prefect FlowRunner | Beginning Flow run for example flow 2019 09 06 20 38 53 241 INFO prefect FlowRunner | Starting flow run 2019 09 06 20 38 53 243 INFO prefect TaskRunner | Task Task1 Starting task run gt gt gt flow run executor=DaskExecutor address= <tcp 127 0 0 1 8786> this doesn t 2019 09 06 20 39 52 658 INFO prefect FlowRunner | Beginning Flow run for example flow 2019 09 06 20 39 52 659 INFO prefect FlowRunner | Starting flow run 2019 09 06 20 39 52 758 ERROR prefect FlowRunner | Unexpected error ModuleNotFoundError No module named example flow ```

Chris White · Answer

Hi < Mikhail Akimov> This is most likely caused by the dask worker not having access to `example flow` The way Dask works in a truly distributed environment is by serializing your function in this case the task and sending it to a worker where it is then deserialized if any dependencies the task needs are not also present on the worker the task will not deserialize I m guessing your `example flow` file is not present on your dask workers in an importable way

Mikhail Akimov · Answer

No it s not But when I run the same file with `python example flow py` and the same DaskExecutor it works OK without the file on any of the workers

Mikhail Akimov · Answer

And the tasks are present in the flow object that I imported For what I know it s identical to the flow I get executing example flow directly

Chris White · Answer

in the file you provided you use a local dask executor though inside the ` main ` block which will run everything in the same place

Chris White · Answer

AH i see the difference

Mikhail Akimov · Answer

Changed to ```if name == main flow run executor=prefect engine executors DaskExecutor executor= <tcp 127 0 0 1 8786> ```

Mikhail Akimov · Answer

Runs OK

Chris White · Answer

in the interactive session you did ``` from example flow import flow ``` this means that from dask s perspective your tasks live at `example flow task3` etc

Chris White · Answer

maybe try ``` from example flow import ``` and see if that fixes it

Mikhail Akimov · Answer

No same error

Mikhail Akimov · Answer

Hm I also see this in worker log ```worker 1 | distributed protocol pickle INFO Failed to deserialize b x80 x04 x95j x00 x00 x00 x00 x00 x00 x8c ```

Mikhail Akimov · Answer

and then a huge blob of bytes

Chris White · Answer

so this is getting into the weeds but if you look at ``` flow module task3 module ``` etc you ll see `example flow` so dask sees all of your objects as living inside a module named example flow that it can t find

Chris White · Answer

when you run it from the CLI the associated module for all these objects is actually ` main ` so everything works

Mikhail Akimov · Answer

``` gt gt gt from large example flow import flow gt gt gt flow module prefect core flow ```

Chris White · Answer

what about the individual tasks

Mikhail Akimov · Answer

``` gt gt gt list flow tasks 0 module prefect tasks core function ```

Chris White · Answer

ah I think it s only `x` the instance of `Task1` that will have the file name as its module

Mikhail Akimov · Answer

Yepp you re right

Mikhail Akimov · Answer

So it s pickle trying to import module when unpickling the function

Chris White · Answer

yup exactly

Chris White · Answer

so if that file is present on the workers in an importable way this will start working

Mikhail Akimov · Answer

Well at my scale making my flows tasks importable on workers will override the usefulness of making a universal launcher for flows Too tedious to keep up to date

Mikhail Akimov · Answer

But thanks anyway learned something new about how pickle works smiley

Chris White · Answer

Just as an FYI for you in Prefect Cloud we have a concept of execution environment and one such environment that we used to use a lot is our `DaskKubernetesEnvironment` which spins up a Dask Cluster for each run of your workflow and gives each worker the docker image containing your Flow regardless of whether you care about Cloud this pattern makes these dependency situations a little nicer at the cost of some latency + resource usage for creating a cluster with each flow run

Mikhail Akimov · Answer

I studied the code for some of the environments and storage and honestly I can t figure out what the abstraction does or why it s needed Does it have meaning outside of Cloud context

Mikhail Akimov · Answer

Oh now I found the examples in the unreleased API docs it became clearer slightly smiling face