hi folks say if I want to delete ALL crashed flow runs or filter by some other condition & delete all flow runs how do I do it? now I am only able to get 200 flow runs in python… not even sure how does it decide which 200 setting up filters for flows in the query is virtual impossible for me (FlowRunFilter, FlowRunFilterState, …. a giant mess doesn’t work) please tell me I am doing it all wrong and there is an easy way? prefect 2 + prefect cloud

👋 curious if

retry_on

functionality is going to be added to Prefect 2 tasks as well? I see it was added in prefect 1.2

With prefect 2, where does one go in the UI to see which agents are connected? (I started an agent and want to make sure it's pointed at the right place.)

👋 Hi all I'm evaluating Prefect 2 for a few ML data pipelines in my team and I'm a little confused about what the right way to achieve what I need. This feels like a really basic use-case so I'm sure I'm just missing something obvious. To make things simple, I have an initial data ingestion flow that brings data into S3 (let's call it

ingestion_flow

). A couple of notes about this: •

ingestion_flow

runs every day and to ingests a data partition for that date. The flow is parameterized with the

date

to ingest. • the flow is organized into a few tasks to ingest data from different clients. So the flow just runs a for loop running

client_ingestion_task(client, date)

for each

client

Once ingested this data is useful for a bunch of downstream pipelines. So I'd like to re-use

ingestion_flow

as a starting point for two other flows, say,

training_flow_1

and

training_flow_2

. The way I thought this could be accomplished was by simply running

ingestion_flow

as a subflow at the beginning of both

training_flow_{1,2}

. Obviously I don't want to ingest the same data twice. I created a caching key for

client_ingestion_task

using the

client

and

date

parameters. However I quickly stumbled upon the fact that task caches only rely on local storage. I'm running these flows in kubernetes and I was hoping for flows to run concurrently ensuring that ingestion only happens once and the cached status gets re-used in the second flow to run. The fact that task caches are restricted to local storage tells me this is not the way to implement what I wanted. Can someone point me in the right direction here? Happy to discuss more details if that's needed.

Hi, Is there a way to cancel all running flow runs of a flow from the Prefect 1.0 UI? Thank you.

I find myself running scripts that set 'stuck' flow runs to Cancelled or Crashed.. Is there some more automatic way to recover from that? It can cause jams, because stuck jobs eat up queue concurrency. Let me know if Discourse or elsewhere is the right place to ask. I also found this Github issue.. how do people handle this in cases of crashed Kubernetes or ECS tasks?

I was trying to use the prefect-aws bundle to upload into the S3 Bucket. I see the S3Bucket object has a

basepath

option but, s3_upload has no basepath.. How do we associate a basepath to the s3_upload option Essentially I don't want the files to land in the bucket root

For those interested on trying Prefect 2, here's a quick repository a did to follow the prefect tutorial more easily on your computer. Simply do a

git clone <https://github.com/jmprovencher/prefect-tutorial>

and follow the steps in

README.md

and

TUTORIAL.md

howdy! Is there a way to kick off a flow run via the prefect libs like outside of a flow

Anyone else loving the dark mode on Prefect Docs? 🙌 🌃 https://prefect-community.slack.com/archives/C03D12VV4NN/p1668187885409289

Getting this error with the fivetran syncs when I move

await

from the function, any ideas?

A 'sync_compatible' method was called from a context that was previously async but is now sync. The sync call must be changed to run in a worker thread to support sending the coroutine for 'load' to the main thread.

Hey Prefect - I have one agent and one queue, I ran a loop over some parameters to break up some work and called

run_deployment

to essentially launch 30 concurrent ecs tasks. What I notice is that 5 of the deployment runs start and the other 25 are pending. How do I make this behave similar to prefect 1.0 where I was able to have all of these run at the same time ? Is this a limitation of my agent ? I was under the impression all the agent does is orchestrate the ecs tasks in my aws environment?

Hello Prefect! I followed the contributing page (newest prefect 2.6.8), but when I run "prefect dev start", I got some error like "AttributeError: 'ArgumentInfo' object has no attribute 'replace'"

Hey folks, I just had a flow run failing because of "prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API." This flow runs several times per day and always succeeded before and also after this incident. The good thing for me: The flow entered failed state and als the logs show an exception - but the flow itself continued after the exception and all the "results" of the pipeline were achieved. Looking at the Exception (in the Thread), is there something I'm missing, or is this maybe something for a github issue? (prefect 2.6.6, prefect cloud)

Hi, I am testing logging from

DaskTaskRunner

workers with the current main-branch version of Prefect, which works now locally thanks to some very recent changes (thanks a lot 💐). However, I am having troubles getting the logs from tasks which are running on a

DaskTaskRunner

which uses

dask_jobqueue.SLURMCluster

. The logs from tasks are written into the slurm-output file, but not propagated back to the flow-log or the cloud UI. Happy, to test some things if it would be helpful.

Is there an issue with the cloud API today? I have agent running which has suddenly (since ~9.20 UTC) stopped picking up flow runs - the agent is up and running - it did have the following error (but killed the agent and restarted since)

WARNING:urllib3.connectionpool:Retrying (Retry(total=5, connect=None, read=None, redirect=None, status=None)) after connection broken by 'ReadTimeoutError("HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Read timed out. (read timeout=15)")': /

Hi Everyone! I need your help . I want to run my code as mentioned below: 1. Expectations: It takes 3 Parameters config_name,table_name,table_key and when I pass it to get_config(get_config is a Prefect Task) it should fetch some data from dynamoDb then pass it to run_config_processor(also a prefect Task) 2. Actual: It is calling run_config_processor without getting the data from get_config. It is running fine on my local.

Hi. We are trying new agent flag released in Prefect 2.6.8 -

prefetch-seconds

. It doesn't work like we expected. Details in the thread.

Hello, thank you for the amazing work on Prefect! I'm wondering whether there's a simple way to set the logLevel for all tasks and flows within Python. Specifically, I'm seeking a way to do this that doesn't leverage logging configuration files or use

get_run_logger().setLevel(level)

within each task and flow. Would this require overriding flow and task initiation, or similar? Thank you in advance for any help!

Please help https://prefect-community.slack.com/archives/CL09KU1K7/p1668775837419239

Hi, our prefect service fails to change status of the flow to failed when container dies or the whole agent dies. (it is prefect 1)

As suggested by Anna, I've looked into her

dataflow-ops

repo and have tried to use the

ECSTask

from

prefect-aws

to circumvent the issue surrounding

DockerRegistry

for ECR, but I keep getting the error

KeyError: "No class found for dispatch key 'ecs-task' in registry for type 'Block'"

. Prefect==2.6.8 & prefect-aws==0.1.8

Hi everyone, is there an example of how to manually set the state of a flow or task to failed in prefect 2?

Hi, is there a way to RETRY the task in Prefect Orion like in the Prefect 1? Thanks

I have a sub_flow that needs to be called by several parent flows. I can do this programmatically and treat the flow as a module and import into each parent flow, but, is there a way I can leverage prefect to do this? For example, can I 'register' a sub flow in prefect and then call it from a parent flow, where the subflow would run within the parent flows infrastructure (not it's own k8s job)? I see that I can use a deployment to deploy the sub flow with the

process

infrastructure setting, but then the sub flow is run by an agent, not the parent flows infrastructure.

Hi can someone help me understand what the arg on triggers are used for i.e. upstream_states. For example,

*prefect.triggers.all_finished*(_upstream_states_)

. This is for prefect 1 btw.

Dear All, I am relatively new to prefect. I am using prefect 2.0. I have followed the official Docs on how to create and run deployments. I am using AWS S3 as my block storage. I successfully upload the files to my block in S3 bucket and the first run is always successful. However, when i try to run the deployment from prefect UI, I always get this error "Flow run <run-name> - Flow could not be retrieved from deployment" AND "TypeError: object NoneType can't be used in 'await' expression". I have tried Docker infrastructure and the default LocalProcess. I have changed the IAM role on the AWS EC2 instance am using but does not seem to work. I NEED HELP. THANK YOU

This is the example am running

I have just started using prefect 2 and nothing seems working for me. I can't start prefect server using prefect orion start. it says background error at: https://sqlalche.me/e/14/e3q8. When I ran the code I get this sqlite3.OperationalError: table flow already exists. I have tried resetting database but it gives this error sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) table flow already exists [SQL: CREATE TABLE flow ( id CHAR(36) DEFAULT ( ( lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2))) || '-4' || substr(lower(hex(randomblob(2))),2) || '-' || substr('89ab',abs(random()) % 4 + 1, 1) || substr(lower(hex(randomblob(2))),2) || '-' || lower(hex(randomblob(6))) ) ) NOT NULL, created DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f000', 'now')) NOT NULL, updated DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%f000', 'now')) NOT NULL, name VARCHAR NOT NULL, tags JSON DEFAULT '[]' NOT NULL, CONSTRAINT pk_flow PRIMARY KEY (id), CONSTRAINT uq_flow__name UNIQUE (name) ) ] (Background on this error at: https://sqlalche.me/e/14/e3q8) An exception occurred. can someone help me with this? I am using ubutu 20.04

I used prefect1 for a year or so, now come back to it all a year later trying to use 2. It is not a very good experience when 99% of all we searches seem to go to v1 info.