Side question: is this still the official Slack for Prefect after moving to 2.0? I was here before but not sure where to ask those questions after we migrate 🙂

When flows crash (according to their logs), Orion UI keeps showing the flows as running, and their run time keeps increasing. If the flow isn't running on the agent, shouldn't it show as "Failed" and not running? Still looking into what actually causes the crash - this is more about how Prefect handles them. We do have flows that runs successful as well, this is just a single flow or two crashing every time.

Hi does anyone know in Prefect 2 why there would be a 30 second delay between the agent picking up a flow run and starting to execute it? Many thanks.

Hi, is there any way to add description of a task in schematic (flow visualisation)

Hi , is there a way to deploy multiple data pipelines(one for daily and other pipeline for weekly etc) , would appreciate the help , thanks in advance

Hi. What I should use instead of wait_for_flow_run(Prefect 1) task in Prefect 2?

In Prefect Live at 3PM Eastern today (Nov 9th), we will host @ben from Club 42. We will talk about how he uses Prefect to coordinate workflows in his college and his experience transitioning from Prefect 1.0 to 2.0. Come join us live on Twitch.

Hi guys, Is there a way to import a function form another file as a task? I was trying something like this:

from tasks.date import get_start_date
@task(name='Generate start date', retries=2, retry_delay_seconds=10)(get_start_date)

It fails with `RuntimeError: Tasks cannot be run outside of a flow. To call the underlying task function outside of a flow use

task.fn()

.` Thanks!

There seems to be a problem with Prefect v1 where new schedules are not being added to flows. Is there an outage on our side?

Can anyone point me towards documentation on setting up a profile to attach to the cloud API from your local machine? Thank you

What's the best way to define dependencies in 2.0?

Hello! I am using Prefect 1.x and store my flows using the Git storage option. I’m trying to “load additional files” from the repo using the steps from the storage docs under Loading Additional Files with Git Storage. However, the file I am looking for isn’t on the file system and I get a “FileNotFoundError: [Errno 2] No such file or directory” error. Looking at the source code for Prefect 1.4.0, the Git storage implementation uses a ‘TemporaryGitRepo’ object to get the flow definition - see line 148 here https://github.com/PrefectHQ/prefect/blob/1.4.0/src/prefect/storage/git.py#L148) which seems like it deletes the repo contents from the file system after reading the flow definition file. I think I’m following the docs correctly, but I might be missing something. I’d appreciate any help and/or suggestions on what to try. Thanks in advance!

Hi! What’s the best way to see error logs or otherwise debug an MS Teams notification? Running Prefect 2.6.4 and I see notification webhooks coming across for flow

Running

states but not for

Failed

— details in thread:

I was able to setup a prefect agent in kubernetes (GKE), and now I'm a little confused on next steps. If you are to create a flow deployment, does that code deployment need to be in a storage block? I'm assuming it will have to be. I tried a flow run without the storage block and got an error saying it couldn't find the flow (and it referenced my local computer's path).

Also, is it possible to keep the agent separate from all the code dependencies by running flows in Docker containers and building all the dependencies within the Docker container? If so, is there a tutorial fo rthis?

Returning Manual State from a Task I want to run a scheduled flow that deals with an intermittent service, and I want to handle the exceptions gracefully. The common idiom seems to be raising

ValueError

to trigger a failure, and thats how to manage the state of subsequent tasks or flows. Here's the thing: I don't want to see a stack trace unless it is some new failure that I need to troubleshoot. For the expected exceptions I want to post an error the log and proceed with my flow according to logic I design. I'm having trouble returning a manual state of Cancelled, concrete example and logs posted in thread. Can anyone guide me toward causing the entire flow to be cancelled without raising an exception?

is there a way to execute long sql queries from a .sql file with sqlalchemy_execute ?

Can we get a /health + /stats endpoint on an agent so we can monitor and cycle if needed

Hi, am struggling to find a standard way to handle flow state changes in Prefect 2. the state handler approach in v1 is deprecated. I need to define a decorator I assume which applies the flow and then applies some kind of state handler on top of that. Does anyone have some code samples on how to do that please?

Hey guys, I wanted to know how I can get a flow to run on multiple schedules with different labels in prefect 1 (this is going to solve my original problem: How can I make a flow to run on every agent at the same time. so if there is a better solution please let me know.)

Hi. Is it possible in Prefect 2 to name the task based on input parameters? https://docs-v1.prefect.io/core/idioms/task-run-names.html

Hi. There is the following suggestion in the prefect-dask documentation:

Then, register to view the block on Prefect Cloud:

Copy code

prefect block register -m prefect_dask.credentials

But there is no module

credentials

in

prefect_dask

. And I get the following error (not a surpirse):

```Unable to load prefect_dask.credentials. Please make sure the module is installed in your current

environment.```

I wonder if this module has ever existed? Or maybe there are plans to create it?

Hi everyone. I am working on a flow on prefect 1.4 and i am struggling to understand why my environment variables are not passing from one task to another. The flow is as follow: • In an upstream task, I add a couple of environment variable to the environment (os.environ object). I logged the object at the end of the task. and I can see that my new environment variables are loaded • Then in a downstream task, I use one of the environment variable and I get a key error. when logging the os.environ object, none of my new environment variable are there. I am missing something? I will add a screenshot of the schematic in the thread.

Hi again, in Prefect Cloud, once you have registered a new block, how does one remove it? So far I can't find anything in the documentation.

I am testing deploying flow to S3 using storage block in prefect 2.0. When I start running flow using Prefect Cloud UI. The agent acknowledged flow run submitted, but wait for 3+ minutes before it starts executing flow run. Does the agent download code in S3 before it starts the execute flow run?

Hi all! What's the best way to implement the ELT flows in Prefect V2? I'm done with EL, I have a sensor task that wait for S3 file and then begins a sync with Airbyte in a Flow. I have multiple file and each file has his own connection, I made a deployment per File, now I want to execute a dbt job after all EL deployments has finished. I'm not sure how to wait for all EL flows, I think that I have to use two subflows one for EL and other for T, but I'm not sure how to pass the params only to one flow or reference the deployment previously created.

Hello, I am trying to use the ECS block to deploy flows on Fargate. I have some questions on this: • Is there a reason to use Task Definition vs. Task Definition arn field in the ECS block? Is it just preference on providing name vs. arn? • Let's say I configure the ECS block with an existing task definition. This task definition includes the ECR image URI which was built with the flow script. But when building the deployment, it looks like I still need to include the path to the flow script. Referencing the example from this doc. So what role is the image playing here vs. reference to the flow script?

prefect deployment build -n prod -q prod -a 
-ib ecs-task/prod -sb s3/prod flow_script.py:flow_function_name

Hello again! I think i’ve tracked down an issue with MS Teams notifications and the notification queue in general where only certain flow run states were being sent. This logger message appears:

prefect-orion-1  | 21:53:34.590 | DEBUG   | prefect.orion.services.flowrunnotifications - Got 3 notifications from queue.

However, it looks like the code intends for only one notification to be read off the queue in the first place: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/orion/services/flow_run_notifications.py#L38-L42.

Hey guys, I’m new to prefect development. Our team is still on v 1.3 How can I fail a task if the task log contains an error? Had a couple flows marked “Success”, but they actually failed in a certain task, the actual errors are recorded in the task’s log and are ERROR level but don’t seem to fail the task.