Hi All, I have a 2.0 Flow/Tasks architecture question: My use case is to Check for one or more files in a s3 buckets, if there is something then execute a few tasks. This check is going to run every 5 minutes but will probably process files only once an hour on average: I separated my code/flows/tasks this way: • A flow named "Check for files on S3) with one task downloading the files and returning the list of downloaded files • If the list is not empty Subflow is called invoked with the list of files that were just downloaded ◦ A few tasks are executed doing transformations and calling some APIs ◦ My last task is to delete from S3 the files I just processed if all of the previous tasks succeeded The reason I did a subflow is so that I can filter easily the flow that is actually processing files instead of having a successful flow every 5min that may not contain any transformation. Is this an antipattern? Also I have to add a

# type: ignore

on every task call because the task returns the type of the wrapped function so I cannot do .wait() on it without raising an issue with mypy. Not sure how we can do better though

Hey, I've encountered a problem with prefect 2.0b12 (I'm running on prefect cloud 2.0). I changed my flow_runner configurations in my deployments to infrastructure. I'm using yaml-configuration for deployments. When trying to create the deployment with prefect deployment create <my-path>, I get the following error:

Found 1 invalid deployments:
    1 validation error for deployment with name 'sync_goodread_score_to_volat_bigquery'
    infrastructure
      Can't instantiate abstract class Infrastructure with abstract methods preview, run (type=type_error)

Please find my deployment in the thread

The latest version of Orion introduces

.submit()

, eg:

@flow
def greetings(names: List[str]) -> None:
    for name in names:
        say_hello.submit(name)
        say_goodbye.submit(name)

but the flow still works without using

.submit()

. What are the implications of not using

.submit()

?

hello. can i spec a arg to restart dask worker every hours and scatter the work's in-memory data to cluster before restart the worker ? i found my dask worker will oom when a long execution time, i can't find the mem-leak reason and so i have to restart works by a period of time, but when i restart worker it will loss the in-memory data and it will re-compute them, this waste time a lot and will make many mistakes.

I have a prefect flow which consumes messages from rabbitmq, I want this task to run continuously, what would the schedule be? I am thinking of starting the flow and then include max_retries and retry_delay for the task which consumes from rabbitmq, is there a better way? thank you

Hi, for some reason my deployments disappeared together with my blocks. I tried to reconfigure the GCP storage (which was already configured) but now I get

(finance-lGLYToYy-py3.9) ➜  finance git:(master) ✗ prefect storage create
No storage types are available. 
(finance-lGLYToYy-py3.9) ➜  finance git:(master) ✗ prefect version
Version:             2.0b10
API version:         0.7.0
Python version:      3.9.10
Git commit:          e71ef86c

Any idea what changed?

Hi, For prefect 2 I understand that you can set notifications to slack. And to email in the cloud version. Will email be available in the non-cloud version as well? And will you be able to create custom events based on state changes of flow runs? Say create Jira ticket when a flow run fails?

Hi Community, I’m using a flow of flow framework, and use KubernetesRun to run the flow. My rough design is like the screenshot: 1. A, B, C, D, E, F are all flows 2. C, D and E, F needs parallel running (where I will go with LocalDaskExecutor), where they basically are running some dbt jobs to do data transformation 3. A, B does not need parallel 4. currently I decide to use a parent flow (as an orchestrator) to run these 6 flows and submit to K8s My questions are : 1. where can I pass parameters (I know I can pass a dictionary params to flow.run() when local run, but no idea with

flow.run_config = KubernetesRun()

) 2. Where can I attach LocalDaskExecutor? Only to the parent flow? (Maybe for flow A, B, I just set their parameters using

unmapped()

function?) 3. Is there a way for me to test my project locally (to make sure the design works as wished) without registering every flow to the Cloud? 4. (Optional) for my use case, is there a better design to achieve my goals? Since I’m quite new to Prefect, but really astonished by its cool functionality, maybe there is some place I can make great improvements for our company Thanks! It’s a long question, but really appreciate for any help!

Hi Prefect Team, I had ask this question in past and asking it again if there is any plan to support caching based on the file target in Prefect Orion as it was supported in Prefect Version 1.0

howdy 🙋 I have a couple of questions regarding best-practices with Prefect. • the first one is regarding a “validation” step (e.g. - if current state in the world already == desired state, then just SKIP the flow entirely). I know i can SKIP a task (by raising skip), but if i wanna skip the entire flow, should i just make sure that the downstream flows depend on it and that’s it? • the second Q is more general about orchestration : let’s say i have an external resource that i want to change to state “GREEN” at some time in the day (every day) and then set back to “BLUE” at some later time in the day (similarly, every day) - in the beginning i thought of doing like a single flow (that would run for example once an hour) and would contain an internal representation the desired schedule - and apply it within the run itself to decide if the state needs to be changed, and if so - to what. Then i figured it might be simple to just run the flow with two different parameters - twice a day (once to make things “Green”, and then again to change them back to “Blue”, with a flow-run parameter — is that even possible in prefect? to have two schedules with different parameters?) --- i’m not sure which one is deemed more “correct”?

hi Community, I have 2 flows: A and B, their dependency is

A -> B

, both A and B will take a list of same org_ids as parameters (need to use map function for parallel running) when I use a parent flow to schedule flow A and B, can I use map function to pass parameters like this way? or is there a better way to do that? code in the thread

Hello, I updated to Prefect2.0 I am trying to run the below commands to start the prefect server but on running the below commands I am getting the error message as below: prefect backend server prefect server start

Hi, I think I must be missing something. I am working my way through the getting started for prefect 2.0 and I got to this page https://orion-docs.prefect.io/tutorials/deployments/#configuring-storage . It seems that the storage command does not exist. Or is this just a windows issue?

In latest Prefect 2.0 beta 2.0b12 deployments registered to backend are not showing for me in the UI. They are showing however when I run

prefect deployment ls

. I tried a reset on orion's database and running the deployment create script again with no luck. This happened after upgrading from b11 to b12

hello prefect team, i'm experiencing errors when trying to execute async flow deployments locally in prefect 2.0. how should i write my async flows so that i local execution works properly? running the flow script locally or running the deployment via an agent works perfectly fine but

prefect deployment execute ...

fails with

RuntimeWarning: coroutine 'create_then_begin_flow_run' was never awaited

. see thread for my flow code. thanks a lot for your support.

This message was deleted.

Hi Prefect team! Since upgrading to Prefect-2.0b12 the module flow_runners no longer exists. Where are we supposed to import e.g. SubprocessFlowRunner from? Have tried to find it in the package but no luck. Thanks!

Hello, i was wondering if we can apply prefect 2 in Object oriented programmation with python . Like having class functions being tasks and flow using objects and those class functions

hello guys, just wanted to give you some feedback on Prefect 2.0. I’m using Pycharm for development, and I’ve noticed that with a lot of Prefect’s classes (like

Deployment

) Pycharm can not provide autocompletion suggestions for the class arguments (like

name

,

tags

…). I think this is because of the decorators applied to the classes which “mask” the arguments (like

@PrefectObjectRegistry.register_instances

). To improve the development experience it could be nice to find a way to expose the arguments so Pycharm and IDEs can provide autosuggestions

Think there may be a bug in the GQL API;

Hi. We use KubernetesRun and see debug logs like this in Prefect Cloud Logs:

Event: 'FailedScheduling' on pod 'prefect-job-79c82cd5-w6nfd'
	Message: 0/7 nodes are available: 2 node(s) had taint {<http://dask.corp.com/component|dask.corp.com/component>: scheduler}, that the pod didn't tolerate, 5 node(s) didn't match Pod's node affinity/selector

Do you have ideas how to disable them?

I can't seem to get tasks running concurrently or in parallel with dask in prefect 2 without using task.submit() Is that expected behaviour? It doesn't seem to be reflected in the docs.

Hello 🙋 is there a default limit for concurrent flows or something in the Cloud? we have currently a flow that was run ad-hoc and is still running, and a flow that was scheduled to run now, and it’s stuck in

Submitted

and nothing seems to be happening with it

Hi all, I am trying to run a flow with VertexRun. I created a GitLab repo, configured the Gitlab storage. Run a agent locally with vertex and the prefect cloud api key. However on run I keep getting the same error. Failed to load and execute flow run: ModuleNotFoundError. It keeps refering to a local path.

Why do I keep seeing this error while I am using the gitlab storage...

Hi prefect team, I had a question about deployment pausing in prefect 2.0. I am little confused by this concept. If a work queue is filtering by deployment and the deployment is paused do all flow runs based on said deployment get paused and go into a "pending" state? I don't see a way "pause" deployment programatically via either the cli or the api. It seems like this is a UI only feature am i missing something here?

Shameless plug: I made a meta-package linking different packages for different use cases. https://github.com/prassanna-ravishankar/python-productivity-stack Under the

mlops

section, I've added prefect -

pip install productivity-stack[mlops]

Hi, anyone knows how can I send params which contain mapped and unmapped parameters to create_flow_run? (More in the thread)

Hey all. Our flow isn't able to spin up more than 180 jobs. We can't determine the root cause exactly but at a certain point there's a connection aborted and prefect starts to remove jobs and then marks the entire flow as failed. We are using a daskexecutor on the cloud. Looking into the dask repos, we aren't able to set some debug level logging as an attempt to retrieve some info on what’s really causing the

Connection aborted

issue. Even if there are some failures, we do want the flow to continue and the task triggers are set accordingly. Thank you

Hi all, does anyone have experience integrating prefect alerts/failures with opsgenie?