Hey guys, Im facing this issue, when trying to install

pip3 install 'prefect[aws]'

WARNING: prefect 2.7.9 does not provide the extra 'aws'

Can anyone help me with this?

Hi folks, I'm having trouble getting my remote worker VM to pick up the code for a deployment I created. I created a flow, then in python, did a Deployment, shown in the second screencap here. I can run this thing on an agent on my machine, but when I try to run it from an agent on my VM, it seems unable to retrieve the file from gitlab. I've got the SSH key and everything set up. I think my problem is the storage block, but I can't quite see what I'm doing wrong because there aren't many examples in the docs about this particular thing

Okay I’ve got a very weird issue for everyone! How do I make a flow finish??? As in, the tasks are done, but the flow is refusing to stop. I’ve got a deployment which executes a single task (that is just a

return 3

) and it runs perfectly…. but the flow continues on and keeps checking for the task, acknowledges its already completed…. but never actually terminates? If anyone has any ideas or has seen this before, please let me know! Note: I’m only seeing issues if using the dask task runners. If it all runs in process, no issue, which leads me to believe theres something configured wrong in my dask cluster. Does anyone have know of instructions on what should be configured (ports open, etc) for prefect and dask to play together? EDIT: I think this might actually be an issue with dark workers and ports, and some prefect is interpreting some of the weird info coming back in a way that makes you think everythings finished when it might have failed

Hi everyone - I need some help. I have somehow created a flow and deployment that aren't required. I want to delete them but for some reason, I can't. I can't delete deployment or flow. I have both

deployment_id

and

flow_id

. I even tried to do it through

prefect_api

- Still it throws errors like

Service Unavailable

when deleting deployment and

Cannot find flow

when deleting flow. Can someone help me with this? UI would just say

failed to delete deployment

Hi all! I’m trying to migrate my Prefect v1 flow to v2 and I’m having a hard time wrapping my head around the process. On v1, we’re using Docker for storage and running them on ECS. How can I do the same in v2? Or do I have to think of this differently? So far I: 1. Created an Prefect agent on ECS using Terraform 2. Wrote my flow code Now I’m trying to deploy it, but how!

Hi! The folder in .prefect is steadily growing in size - is this the default space for the database or is it some temp files that are not deleted?

On prefect 2, what’s the best way to do a flow run without tracking by Prefect Cloud? My

~/.prefect/profiles.toml

is configured to connect to Prefect Cloud, but sometimes I just want to run flows locally without cloud tracking. I was able to figure something out with environment variables, but it seems hacky:

import os

# To disable tracking by Prefect UI, set these environment variables to empty values.
os.environ['PREFECT_API_URL'] = ''
os.environ['PREFECT_API_KEY'] = ''
os.environ['PREFECT_PROFILES_PATH'] = 'dont' # set to a path that doesn't exist


import prefect
import prefect.settings

from prefect import flow, get_run_logger

@flow
def test_run():
    logger = get_run_logger()
    st = prefect.settings.get_current_settings().dict()
    keys = sorted(st.keys())
    for k in keys:
        v = st[k]
        <http://logger.info|logger.info>(f"{k}: {v}")

# This run won't be tracked by Prefect Cloud
test_run()

Hi, can anyone tell me if prefect 1 keeps task results in memory for the duration of the flow? Especially with DaskExecutor? We have a flow that handles some large outputs/inputs at and early stage in the flow, but they have no utility for the end result of the flow. It seems like they might be held in memory throughout the entire duration of the flow. I notice there is an option

cache_result_in_memory=False

in prefect 2, but does anybody know if there is a way to disable memory caching in prefect 1?

hi there, i'm seeing an error:

RuntimeError: <asyncio.locks.Event object at 0x11447f760 [unset]> is bound to a different event loop

when attempting to run subflows in parallel, but only when running against Prefect Cloud. details in thread

How do I get the flow run id from context? (not the flow id) (prefect 2)

What's the recommended way to autodetect flows? Let's say I have a bunch of flows in a

flows

module, would it still be to build a deployment for each of them?

Is there any documentation available on what each section of the Custom Triggers json for automations controls? For instance I’m looking to set up an automation that would fire a notification if deployments with a specific tag fail 4 times in a row, but it’s not clear to me how exactly the fields need to be set to accomplish that.

What is the format for adding multiple pip packages for a KubernetesJob?

How should web applications (Node.js) listen/subscribe to Prefect workflows completion? I have a task that trains a deep learning model through a Web Application. The “*Start training*” button on the web app sends a

POST /model

REST API call to the Node.js server which then in turn calls Prefect 2.0 using the REST API to start a new run (of an existing deployment). The workflow will train the model, run inference on some data and then persist the output to a database. This is obviously an asynchronous operation that may take a few minutes (or more) to complete. Assuming the workflow succeeded, I would like to notify the users (those who have Chrome opened to my web app) that something has happened, i.e. training completed. How should the Node.js be notified when the flow has finished (either successfully / failed)? Is there like a middleware / RabbitMQ / other message queue that the Node.js app can subscribe to, onto which Prefect publishes event? If not, does Prefect expose other broadcast events? And if not, should I poll periodically from my app and maintain state diff? Thanks!

How many concerns should one workflow deal with? This is a system design question. Let’s say that I have a web application that persists its data to its own PostgreSQL Docker instance. Then I have my workflow deployment through the Prefect orchestrator, and that workflow writes some data to, say, Snowflake. Is it OK for my workflow to deal with cross cutting concerns such as updating both the Snowflake (business logic data model) and the PostgreSQL (for the web application data model)? Or should the web app (e.g. Node.js) be subscribed, notified and act upon workflow completions? I can view the workflow as a conductor that knows all the moving parts of the system, e.g. all different databases, but I wanted to hear from real usages (in the wild) about system design considerations and separation (or mixing) of concerns and what are the tradeoffs/considerations. Thanks!

Hello, team! I'm using prefect 1.2.1 and using below code to register Flow

with Flow("floww_3.5", storage=Azure(container="prefect-flows")) as flow:
    input_folder_path = Parameter("input_folder_path", default='demo_path')
    utput_folder_path = Parameter("output_folder_path", default='')
    .......

if __name__ == "__main__":
    client = Client()
    client.create_project(project_name="floww_3")

    project_path = str(Path.cwd())
    flow.run_config = KubernetesRun(.....)
    flow.executor = LocalDaskExecutor()
    flow.register(add_default_labels=False, project_name="floww_3")

Every time I execute above code it's getting registered properly without any error and version is also getting upgraded. But if I change default values of Parameter it's not getting updated Could someone please help with it?

This message was deleted.

Hi! I am facing an issue with flows crashing due to:

prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

I am running same deployment multiple times with different parameters via API call using python requests. 4/5 runs of the deployment failed with this error. They start running and crashed half way. Prefect 2.8.2, prefect-aws 0.2.4. My flows are running in ECSTasks. Any ideas why this happens?

azure docker ship Docker infrastructure block tries to access flow file from local system instead of azure storage Hi all, I am currently running a prefect cloud, with azure blob storage as storage block and a docker container from a private repository as an infrastructure block. The issue that I am facing when running the flow. The agent that I am running on my local machine pulls the docker image and runs it (confirmed from the container logs in docker desktop) before throwing the following error and exiting:

03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/engine.py", line 246, in retrieve_flow_then_begin_flow_run
2023-03-03 13:24:50     flow = await load_flow_from_flow_run(flow_run, client=client)
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/client.py", line 105, in with_injected_client
2023-03-03 13:24:50     return await fn(*args, **kwargs)
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/deployments.py", line 73, in load_flow_from_flow_run
2023-03-03 13:24:50     flow = await run_sync_in_worker_thread(import_object, str(import_path))
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/utilities/asyncutils.py", line 56, in run_sync_in_worker_thread
2023-03-03 13:24:50     return await anyio.to_thread.run_sync(call, cancellable=True)
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/anyio/to_thread.py", line 32, in run_sync
2023-03-03 13:24:50     func, *args, cancellable=cancellable, limiter=limiter
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
2023-03-03 13:24:50     return await future
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/anyio/_backends/_asyncio.py", line 867, in run
2023-03-03 13:24:50     result = context.run(func, *args)
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/utilities/importtools.py", line 193, in import_object
2023-03-03 13:24:50     module = load_script_as_module(script_path)
2023-03-03 13:24:50   File "/usr/local/lib/python3.7/site-packages/prefect/utilities/importtools.py", line 156, in load_script_as_module
2023-03-03 13:24:50     raise ScriptError(user_exc=exc, path=path) from exc
2023-03-03 13:24:50 prefect.exceptions.ScriptError: Script at '<Flow script location on local machine>' encountered an exception

This seems related to this issue however as I am running prefect 2.1.0

DockerFlowRunner

has been deprecated. I tried the

COPY $FLOWSCRIPT.py .

before building the docker image as well but it didn't help, and isn't a solution that will work for me. The link in there to the s3 with volume parameters for the credentials also doesn't seem relevant, as I have already loaded and authorised the storage block using SAS. The same issue occurs even when I deploy using the CLI according to the prompt in the second to last row of this table. Could someone please help me to get the docker container to look for the flow script in the storage block where it is successfully uploaded on deployment, not the path of the flow script on the machine that initiates the deployment?

Hopefully a simple question: In the definition of task (https://github.com/PrefectHQ/prefect/blob/main/src/prefect/tasks.py) it doesnt give any explanation of what cache_result_in_memory means. Can someone briefly explain what happens when that is True and what are the possibly implications if set to False?

Hello everyone, we are running several flows in Prefect v1. We are using k8s with Dask workers. We recently noticed an issue where some tasks are executed multiple times even though first execution succeeded. We believe this happens (mostly) when a Dask worker dies. We also found an open issue that confirms the conditions for this to happen: https://github.com/PrefectHQ/prefect/issues/5485 I understand there is less appetite to working on Prefect1 with v2 now out and being utilized but since we are not ready to make the migration right now, I was hoping someone can give us some more guidance on how we can possibly mitigate this issue. Thank you!!

I'm curious if there's a way to automatically remove flows from our queue if they are late? We have a flow that runs every 3 minutes and if our agent goes down, we end up having hundreds of flows that all start running when the agent is restarted.

Would anyone know why an azure function isn't able to connect to the prefect cloud? I set the correct api_url and api_key in the az function config settings and it all worked well about a month ago but now I'm getting an error of what looks to be from prefect that the cloud request isn't returning anything.

Can you customize the logging format of the Prefect Agent? I put the logs into Datadog and i’d rather not do a bunch of parsing. Structured logging would be fantastic

How odd is it to register a schedule for a workflow from a different running workflow? We have an application where users trigger training of ML models through a web app. Upon triggering that, we would want to register a daily scheduled workflow to be run every day. Does Prefect allow to express that? is there best practices around that?

Currently I’m using prefect==2.7.4 and I plan to migrate to the latest version (2.8.4). Is it ok to first migrate the prefect server and later on the agent? Or do you strongly recommend always migrating both together? E.g. I’ll might have my system running like this for a while: agents==2.7.4, server==2.8.4.

I've just discovered the

allow_failure

utility - is there something similar to invoke only when an upstream task fails? My use case is a flow running dbt models, I want to run failed models when the original task fails

Hey all, I just got started with prefect and like the workflow. My question, does prefect-sqlalchmy support inserto into select queries? I am trying to execute an 'insert into select columns from table' query which works when I interact with the db directly, but not through the SqlAlchemyConnector.execute(query_string) method. Wasn't sure if it wasn't supported or I was doing something wrong

anyone ever get the 'Can't connect to Server API at http://127.0.0.1:4200/api. Check that it's accessible from your machine.'? After developing locally, I'm setting up a remote host for actual deployments and unable to get the server configured correctly it seems. I'm starting the server with:

prefect server start --host=0.0.0.0

hey, is the “Projects” abstraction (as in Prefect 1) on the roadmap for Prefect 2?