Say I want to start a second prefect agent (running prefect 2.0) so we can start developing some flows to run on prefect 2.0, but want all existing flows to continue running via the prefect 1.x agent. Is there a way to do that without needing to change the labels on our existing flows?

Hi Team - I tried to use this flow to flow in Prefect 1.0. https://docs-v1.prefect.io/core/idioms/flow-to-flow.html . I ended up getting this error: [2022-09-12 16:55:40-0500] ERROR - prefect.sod-loader-parent-flow | Unexpected error occured in FlowRunner: ValueError('Cycle found; flows must be acyclic!')

Small question (or 2) about prefect 2.0 Trying to build a CICD solution with prefect 2.0. There are a few things I’m still struggling with; • What is the advantage of storing my flow on another source than Git? Using GIT gives me a single source of truth inline with the standards in my organization. • Where does my flow store its cache while running? I want to be able to restart a task when my flow crashed, while running on K8s. Am I correct when assuming that I need to configure External Storage with put-directory and write-path for this? • The MINIO example on the website is trying to connect to an AWS endpoint, probably due to s3fs. Can I just follow the example on https://docs.prefect.io/concepts/blocks/, create my own PyMinio based block and use that? Many thanks in advance

Hi Prefect Team, We're planning to use Prefect for creating our data flows, If I use the open source one, Do I need to host Prefect on my own cloud provider?

Hi! My flow seems to have finished by still show up as running in Prefect Cloud. This is the last logged message in the agent log (in the execution environment)

08:12:06.349 | INFO    | prefect.infrastructure.docker-container - Docker container 'practical-bear' has status 'exited'

But the log on prefect cloud is stuck on Executing of the last task of the flow.

Prefect adjusts the run time at the end of run. Any reason for this? Feels a bit sketchy..

I'm trying to use a shell_run_command to run some soda.io scans again my dbt models with prefect. The execution is correct but if the test itself fails, the tasks returns an exception and so my flow is also failing. Is there anything I can do to prevent the task from terminating in an incorrect state?

09:19:54.856 | INFO    | prefect.agent - Submitting flow run '9e981ee3-5d70-43b8-992a-6bd5c433d24c'
09:19:55.054 | INFO    | prefect.infrastructure.process - Opening process 'blond-rabbit'...
09:19:55.130 | INFO    | prefect.agent - Completed submission of flow run '9e981ee3-5d70-43b8-992a-6bd5c433d24c'
09:19:57.018 | INFO    | Flow run 'blond-rabbit' - Created task run 'shell_run_command-7398b6ba-0' for task 'shell_run_command'
09:19:57.019 | INFO    | Flow run 'blond-rabbit' - Executing 'shell_run_command-7398b6ba-0' immediately...
09:19:57.251 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Soda Core 3.0.6

09:19:58.517 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472472552592 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.518 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Lock 281472472552592 acquired on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.518 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472472682848 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Timeout on acquiring lock 281472472682848 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:19:58.519 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to release lock 281472472552592 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Lock 281472472552592 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.123 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Scan summary:
1/1 check FAILED: 
    v_stg_transactions in we_liberate_data
      row_count between 10 and 1000 [FAILED]

09:20:00.125 | INFO    | Task run 'shell_run_command-7398b6ba-0' -         check_value: 6209
Oops! 1 failures. 0 warnings. 0 errors. 0 pass.

09:20:00.125 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Sending results to Soda Cloud

09:20:00.484 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Soda Cloud Trace: 3791758754192003542

09:20:00.878 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to acquire lock 281472486055312 on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.878 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Lock 281472486055312 acquired on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.879 | INFO    | Task run 'shell_run_command-7398b6ba-0' - Attempting to release lock 281472486055312 on /home/vscode/.cache/snowflake/ocsp_cache.lock
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.930 | ERROR   | Task run 'shell_run_command-7398b6ba-0' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect_shell/commands.py", line 90, in shell_run_command
    raise RuntimeError(msg)
RuntimeError: Command failed with exit code 2:
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.952 | ERROR   | Task run 'shell_run_command-7398b6ba-0' - Finished in state Failed('Task run encountered an exception.')
09:20:00.952 | ERROR   | Flow run 'blond-rabbit' - Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 588, in orchestrate_flow_run
    result = await flow_call()
  File "flows/test/test_soda_transactions.py", line 11, in test_transactions
    command2 = await shell_run_command(
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 863, in get_task_call_return_value
    return await future._result()
  File "/usr/local/lib/python3.8/site-packages/prefect/futures.py", line 236, in _result
    return final_state.result(raise_on_failure=raise_on_failure)
  File "/usr/local/lib/python3.8/site-packages/prefect/orion/schemas/states.py", line 143, in result
    raise data
  File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect_shell/commands.py", line 90, in shell_run_command
    raise RuntimeError(msg)
RuntimeError: Command failed with exit code 2:
Lock 281472486055312 released on /home/vscode/.cache/snowflake/ocsp_cache.lock

09:20:00.966 | ERROR   | Flow run 'blond-rabbit' - Finished in state Failed('Flow run encountered an exception.')
09:20:01.240 | INFO    | prefect.infrastructure.process - Process 'blond-rabbit' exited cleanly.

Hello, I need suggestion: In deployment with prefect.flow_runner import subprocess deprecated. How can I set virtualvenv as like in SubprocessFlowRunner(virtualenv=path_to_interpreter) What is possible way I can gracefully upgrade this line of code?

Hi, I'm having some trouble to understand how to better restart subflow tasks, thought to ask for some help :) Here is situation: I have flow which contains subflow tasks. Subflow D has upstream of two subflows A and B. They all have wait_for_flow tasks. If for example subflow A fails, I want in prefect UI restart all subflow D downstream from where it failed. But now if I try to restart in UI it will start to restart subflow A wait_for_flow task without running subflow A task itself and it will be stuck waiting. So my question how to better restart subflow D downstream(including subflow A) without restarting subflow C and B?

Hi, trying out some simple map reduce stuff on latest prefect on a dask cluster. Should I use task.submit() or can I also use task.map()?

Hi, is it possible to get the previous flow run metadata of a deployment when running a new one, besides querying the orion postgres itself? When updating our data we would need a time delta for when the ETL last ran.

This message was deleted.

Also how much memory and CPU will I need to run a prefect agent, given that I’ll be running kubernetes jobs (as separate pods)?

In prefect 1, when starting a local agent from a container, I pass prefect agent local start -k <api key value>. Is doing the same thing possible for prefect 2? Reading the docs, it doesn't look that passing an api key is an option for agent start.

Hey guys, We are running a few pipelines on Prefect 1.2 (on-prem Kubernetes) and recently i've started working on migrating to Orion. We have two use cases that query Prometheus metrics: streaming flows running at seconds interval and batch flows running for the previous day, weeks etc. In Prefect 1.2 I had two projects, one for streaming and one for batch. In this way, I was able to register a flow having the same name in both projects (of course, having different tasks under the hood). For example, i have a flow called 'Alerts' in both projects, one running daily and fetching previous day data and another one running at 30 seconds interval. In Orion, it seems it's not possible anymore to isolate the two use cases. Is there another way? I've read about Workspaces, but that it's available only in Prefect Cloud. Another question is about starting streaming flows. My flow consists in having one task creating a table and another one running a while loop that calls the Prometheus api every 30 seconds and inserts data into the table. So i need to run in streaming mode just one task (the api call), not the entire flow. In Prefect 1.2 i am scheduling the flow to run only once 10 seconds after the registration (using

from prefect.schedules import Schedule

). In this way, it's entering the task while loop and the UI shows only one flow run in "Running" state (until you decide to kill it). Is there a way to replicate this scenario in Orion? If i use the Interval schedule, it's creating flow runs every 30 seconds and it's difficult for the users to read the UI once you have more flows running. Thank you!

Hello all, I'm trying to evaluate if Perfect fits my company's needs. For our use case, we would have a flow that dispatches CPU and GPU tasks. Is it possible to easily switch between the following execution modes: 1. ALL tasks are scheduled to the same remote cluster 2. CPU and GPU tasks are scheduled to a CPU and GPU remote clusters respectively 3. All tasks are executed locally Documentation that I'm finding suggests that there will be always a single task runner per flow, and that the task runner is specified at flow definition time, ie I cannot change execution mode without modifying code.

hey everyone! I’ve been getting HTTP Read Timeouts on my flow execution from Prefect 2.0 API; just checking if something I can do or if this is a usual issue for everyone?

Hi all. Is there any chance state handlers for slack notifications will make it into prefect 2 as task and flow arguments? I’d built a pretty extensive one for myself in prefect 1 and would like to use the same for prefect 2

Hello, I've been running into this error trying to login to prefect cloud via docker image. The Dockerfile is simple: FROM prefecthq/prefect:2.3.2-python3.10 RUN prefect cloud login -k <my api key> I get the following attached error when running docker build . I tried using different prefect images with different prefect2 and python versions, but get the same error. Any idea what could be causing this?

Is prefect 1.0 cloud going to be discontinued?

Also how to keep agent for prefect 2.0 keep on running in background. Apprently as soon as I close my session, all my scheduled jobs go into late mode. not sure why

prefect agent start -q 'default'

Hello, I am unable to run the cythonized prefect code, got the error below:

result = pickle.dumps(x, **dump_kwargs) _pickle.PicklingError: Can't pickle <cyfunction xxxx at 0x7fff2e711390>: it's not the same object as pipeline.xxxx During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner new_state = method(self, state, *args, **kwargs) File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 628, in get_flow_run_state extra_context=extra_context(task), File "/usr/local/lib/python3.7/site-packages/prefect/executors/dask.py", line 422, in submit _ma

any idea? looks like it's unable to pickle prefect task

@task def xxxx(config):

task xxxx is defined in pipeline.py as above

how is

prefect.client.get_client()

supposed to know which account of prefect cloud to use? how does it get the api key?

Hello all, I would like to ask if you could perhaps help me with a problem starting the Prefect Orion server locally, that occurs since version 2.3.0? After excecuting the command

prefect orion start

in windows powershell, the Prefect Orion logo appears with information about the Orion URLs but is immediately followed by the error message:

"Error: Got unexpected extra argument (prefect.orion.api.server:create_app) Orion stopped!"

The problem occurs since version 2.3.0 (it occurs also in 2.3.1 and 2.3.2). Version 2.2.0 works well. My operating system is Windows 10 and I would attach the information from the

prefect version

command into the thread. If you had any idea what could be the cause of the error, you would help me a lot! Thank you very much in advance!