morten
07/25/2023, 10:04 AMJake Kaplan
07/25/2023, 10:48 AMmorten
07/25/2023, 10:53 AMDoes each flow run need to be executing specific dependencies in it's own isolated virtual environment?Well, strictly speaking no, not at the moment. but as my of flows are going to grow, I will likely need some way to ensure that if one requires package A with version 1.0.0 and another requires version 2.0.0, its all possible.
Additionally what worker type are you using?I am starting my worker with
prefect worker start --pool my-process-pool
where the pool is a Process
type. Not sure if that is what you are asking(?)Jake Kaplan
07/25/2023, 12:20 PMcommand
on your work pool? The default command is python -m prefect.engine
but you may be able to change it to source /{venv}/ && python -m prefect.engine
This should also be overridable per deployment that uses that work pool.
Alternatively:
⢠It's not poetry, but one thing you can use is the https://docs.prefect.io/2.11.0/concepts/deployments-ux/#utility-steps pip_install_requirements
pull step. This will install into your current working environment though while the flow run is running (not a new one).
⢠For managing different environments, the "prefect" way to do this is with container based worker/work pools. Aside from the Process
type pretty much all the other worker/work pool types involve transient containers. This gives you the benefit of having an environment specifically for that 1 run of a flow. You can either use the pull step above for each deployment of a flow that has different dependencies or even bake them into your own images if you prefer.morten
07/25/2023, 12:29 PMJake Kaplan
07/25/2023, 12:31 PMCloudRunWorker
can only read from a CloudRunWorkPool
. Both have type "cloud-run"
morten
07/25/2023, 12:31 PMJake Kaplan
07/25/2023, 12:31 PMmorten
07/25/2023, 3:26 PMJake Kaplan
07/25/2023, 3:45 PMmorten
07/25/2023, 3:55 PMJake Kaplan
07/25/2023, 5:05 PMas that would be one less tool I would have to maintain.I definitely can relate to that feeling š were you able to try changing the command to include activating the environment?
morten
07/25/2023, 5:07 PM06:30:36.449 | INFO | prefect.flow_runs.worker - Opening process...
06:30:36.454 | ERROR | prefect.flow_runs.worker - Failed to submit flow run '55bf52ef-755e-4a60-8245-21f34fc16b5b' to infrastructure.
Traceback (most recent call last):
File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 834, in _submit_run_and_capture_errors
result = await self.run(
^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/prefect/workers/process.py", line 176, in run
process = await run_process(
^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/prefect/utilities/processutils.py", line 258, in run_process
async with open_process(
File "/usr/local/lib/python3.11/contextlib.py", line 204, in __aenter__
return await anext(self.gen)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/prefect/utilities/processutils.py", line 202, in open_process
process = await anyio.open_process(command, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/anyio/_core/_subprocesses.py", line 126, in open_process
return await get_asynclib().open_process(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 1041, in open_process
process = await asyncio.create_subprocess_exec(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/subprocess.py", line 221, in create_subprocess_exec
transport, protocol = await loop.subprocess_exec(
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/base_events.py", line 1694, in subprocess_exec
transport = await self._make_subprocess_transport(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/unix_events.py", line 207, in _make_subprocess_transport
transp = _UnixSubprocessTransport(self, protocol, args, shell,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/asyncio/base_subprocess.py", line 36, in __init__
self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
File "/usr/local/lib/python3.11/asyncio/unix_events.py", line 818, in _start
self._proc = subprocess.Popen(
^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.11/subprocess.py", line 1026, in __init__
self._execute_child(args, executable, preexec_fn, close_fds,
File "/usr/local/lib/python3.11/subprocess.py", line 1950, in _execute_child
raise child_exception_type(errno_num, err_msg, err_filename)
FileNotFoundError: [Errno 2] No such file or directory: 'source'
06:30:36.481 | INFO | prefect.flow_runs.worker - Completed submission of flow run '55bf52ef-755e-4a60-8245-21f34fc16b5b'
I have no idea why it tries to think if source as a file or directoryJake Kaplan
07/27/2023, 12:47 AMsource
from inside of a process.popopenpython -m prefect.engine
you would make it use python executable of the virtualenv you want like
/Users/jakekaplan/opt/anaconda3/envs/py310/bin/python3 -m prefect.engine
morten
07/28/2023, 10:18 AMJake Kaplan
07/28/2023, 11:35 AM