Open tekumara opened 2 years ago
When the flow script is run directly, we lose the ability to infer the module name since it's replaced with __main__
if you call python <your_script.py>
or __prefect_loader__
if we run the script for you. If you move your deployment into a separate file and import the flow, we will be able to use that as the flow's import path:
from my_flows import kubes_flow
Deployment(
flow=kubes_flow,
)
I'll try to find a way to determine the module path when they're in the same file.
Also note, if your module is a relative import rather than an installed module, this will fail while using the CLI. I'll investigate a fix for that, but you can call Deployment(...).create()
directly and run the script with Python to get past it.
Thanks @madkinsz I'll try that. FYI I'm creating the deployment via the CLI, eg: prefect deployment create flows/kubes_flow.py
.
Yep splitting out the deployment into its own file works! 🥳
I get a similar error when using DaskTaskRunner
and invoking a function from within the same file as the task and flow.
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/distributed/worker.py", line 2157, in execute
function, args, kwargs = await self._maybe_deserialize_task(ts)
File "/usr/local/lib/python3.9/site-packages/distributed/worker.py", line 2127, in _maybe_deserialize_task
function, args, kwargs = _deserialize(*ts.run_spec)
File "/usr/local/lib/python3.9/site-packages/distributed/worker.py", line 2755, in _deserialize
kwargs = pickle.loads(kwargs)
File "/usr/local/lib/python3.9/site-packages/distributed/protocol/pickle.py", line 73, in loads
return pickle.loads(x)
ModuleNotFoundError: No module named '__prefect_loader__'
I have the following 2 files.
flow.py
:
from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner
def inner():
return 0
@task
def bar():
inner()
@flow(task_runner=DaskTaskRunner())
def foo():
future = bar()
future.wait()
spec.py
:
from prefect.deployments import Deployment
from flow import foo
spec = Deployment(
flow=foo,
tags=["test"],
)
Related to #6629
FWIW I ran into a similar error when attempting to run a computation on a Dask cluster from inside a running Prefect flow. Here's a description and change to distributed
to avoid the particular issue I was running into https://github.com/dask/distributed/pull/8502. Just posting here for visibility.
Description
The flow run errors with:
Reproduction / Example
_flows/kubesflow.py:
Here's the block for my flow:
It looks like
__prefect_loader__.kubes_flow
is being stored as the module name rather thanflows.kubes_flow
prefect 2.0b8