Heya! 👋 What would be the best practice in determining, if flow needs to run the ETL or not, depending on source data being stale or not. Context: I've got a google sheet from which I'd load data into Snowflake, which works great, however I'd like to avoid communicating with Snowflake, if the GS has not been updated since the last successful flow run. My options the way I see it are • Store state somewhere externally (would require quite a bit of extra code and moving parts) • During flow runtime, check previous flow run status and time of run • Check when the GS was last updated and compare the delta between that and NOW() with the flow's scheduled interval retrieved either during runtime or hardcoded

Hey all, I have a Prefect 2 flow using DaskTaskRunner (Reading data from multiple unrelated sources, preprocessing and then training a model). I also use map as part of the flow (for parsing some objects in parallel). Everything works fine until I get this error (related to Dask) -

12:36:08.900 | INFO    | Flow run 'logical-peacock' - Created task run 'calculate_additional_features-0' for task 'calculate_additional_features'
2023-04-30 12:36:08,915 - distributed.protocol.pickle - ERROR - Failed to serialize <ToPickle: HighLevelGraph with 1 layers.

What is the problem here? what am I missing? Sharing my flow code as a comment on this message. Setup - running locally with pycharm’s docker compose interpeter, Prefect 2.10.4 prefect-dask==0.2.3 Thanks.

👋 I don't understand what changed in Prefect 2.0 cloud / agents recently. Everything was running smoothly but now when I start my agent, As soon as I strat a quick flow run, the work pool becomes unhealthy and I get the following error sometimes (most of the times I don't get any error).

does not have an infrastructure pid attached. Cancellation cannot be guaranteed.

It's very hard to have more logs than that so it's hard to understand the problem and debug. Any help would be helpful 🙏 Some more context • The agent is deployed in GCP VM • I use the exact same command line than specified in Cloud • Stopped working last week with Prefect 2.7.* on the agent side. Same problem with 2.10.*

What happens in Prefect when you restart a failed run? e.g. let’s say I had retry=3 but all retries failed because the database was down and the last task in the flow needed to ingest data into the database. One hour later the database comes up and I manually restart a failed run. Does the restarted run know it’s yet another an incarnation of a previously failed run, or is it stateless and think it’s a different run? I guess my question is whether Prefect re-runs are stateful or stateless and what are the re-try semantics? Also what are the best practices around idempotency? I want an all-or-nothing semantics with the final results written to database, no partial leakage. So the question also becomes whether or not re-runs that wrote no data are considered successful. Update: I saw this post from @Anna M Geller https://discourse.prefect.io/t/how-can-i-restart-a-flow-in-prefect-2-0/1332/7 I think there is a hidden “Data Engineering 101” assumption that: 1. Tasks and DAGs as a whole should be idempotent 2. Use timestamps as means to achieving idempotent tasks - e.g. max(LAST_UPDATED), this is the flag that will indicate whether the task already succeeded, so no side effects 3. Implement atomic all-or-nothing semantics within each task a. For example - no partial writes to database tables

There are probably recommended complementary books / courses to Prefect / Airflow, as there are data engineering principles that stick with you regardless of the framework you use to implement your data pipelines. I would love to learn good principles and approaches towards designing and implementing DAGs.

🔥 Sorry for posting this issue here as it’s already posted in Github. It’s just super critical for us, and we are stuck running flows in our production env. When editing an ECS Task Block in the Prefect Cloud UI, when you clear a json field, at won’t come back to a None value but instead set to an empty string “”. There is no way to remove this empty string other than creating a new ECS block. But this new Block can’t have the same name. It would be great if someone from Prefect’s team can have a quick look. Thank you for any help on the issue! 🙏 https://github.com/PrefectHQ/prefect/issues/9378

I can't seem to resolve

from prefect.server.schemas.schedules import CronSchedule

. Is there an additional library that I need to install? Is that only available if I am running the server local? I'm looking to use

Deployment.build_from_flow()

to deploy to the cloud.

Looking for a way to conditionalize task retries. The API I am extracting from can produce a few exceptions, some of which should stop the flow altogether and others that should be retried. I'm looking for the Prefect code that manages retries, but have had little luck. Would returning a

Cancelled

state vs a

Failed

state work?

Hi, any frenchy (or French reader) which would help me to write a blogpost about "how to secure prefect with API key on promise" ?

Hi there, I am looking into producing a table artifact in my flow but am not able to figure out why the artifact is not showing up in the UI for this flow… My flow is set up with a concurrent task runner so I needed to

await

the call to

create_table_artifact

in order to get the flow to run, maybe that could be the issue?

Hi there, I am fairly new to Prefect and trying to use the REST APIs. I have a local server running on local Postgres as the database. I was able to apply a deployment and run a flow using the prefect command line as well as from the UI. Great. I then opened the API docs and successfully ran GETs for both the flow and the deployment. But when I try this: /api/deployments/{id}/create_flow_run Create Flow Run From Deployment (I take the generated curl as before and run it) I see the following error message on the server (I cut out the stack trace): sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class ‘asyncpg.exceptions.ForeignKeyViolationError’>: insert or update on table “flow_run” violates foreign key constraint “fk_flow_run__parent_task_run_id__task_run” DETAIL: Key (parent_task_run_id)=(3fa85f64-5717-4562-b3fc-2c963f66afa6) is not present in table “task_run”. [SQL: INSERT INTO flow_run (id, created, updated, name, run_count, total_run_time, deployment_id, work_queue_name, parameters, idempotency_key, context, empirical_policy, tags, auto_scheduled, flow_id, infrastructure_document_id, parent_task_run_id, work_queue_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (flow_id, idempotency_key) DO NOTHING] [parameters: (UUID(‘67eeff72-9ce7-4120-b85c-63067bbd724a’), DateTime(2023, 5, 1, 17, 23, 7, 149371, tzinfo=Timezone(‘UTC’)), DateTime(2023, 5, 1, 17, 23, 7, 150505, tzinfo=Timezone(‘UTC’)), ‘my-flow-run’, 0, datetime.timedelta(0), UUID(‘4d2be4ab-1fe5-49f3-a6f0-9d7ccf8bb4af’), ‘default’, ‘{}’, ‘string’, ‘{“my_var”: “my_val”}‘, ‘{“max_retries”: 0, “retry_delay_seconds”: 0.0, “retries”: 0, “retry_delay”: 0, “pause_keys”: [“string”], “resuming”: false}‘, ‘[“tag-2”, “tag-1"]’, False, UUID(‘704aec56-b740-4400-8c52-5e08314ec84a’), UUID(‘3fa85f64-5717-4562-b3fc-2c963f66afa6’), UUID(‘3fa85f64-5717-4562-b3fc-2c963f66afa6’), UUID(‘084966f2-e8e3-4827-8638-d4a8642f0f2c’))] (Background on this error at: https://sqlalche.me/e/14/gkpj) I appreciate any insights into what I might be doing incorrectly. The task_run table is empty (should it be)? Thank you!

Hi all! I'm storing my flow code on Github, created a github block, ran the deployment command, and started the agent. However, I'm getting the following error:

...
OSError: Failed to pull from remote:
 Cloning into '/var/folders/cr/8m8l_rj15d1930_z5w0z9qnc0000gn/T/tmp3cmpibx_prefect'...
fatal: repository '<my_github_repo_link_returned_here>' not found

I'm not sure what I could be doing wrong? Anyone to help?

New logo is 🔥

Work Queues becoming unhealthy I have code running on a cloud server. Agent on server is started using

nohup prefect agent start  --work-queue "<name>" > ~/tmp/prefect_agent.log &

It use to work fine, but recently I notices it becomes “unhealthy” and stops running flows. any idea why this might happen and how to prevent this ? thanks

Hello! I’m using prefect deployments to push my modules code to S3 and then pulling it back down into the container but I keep running into a

ModuleNotFoundError

for anything that is defined outside of the flow file. How do I pull in all my other code into the dependency tree?

New to using Prefect...there's gotta be something I'm missing in setting up. Deployed in GCP, using Prefect CLI and dbt and BigQuery. Just trying to write a simple flow for testing. I used the server UI to save a block for the link to my test dbt repo. The repo is public. I'm trying to reference that block in my flow, and I just keep getting an error that the project_dir path is invalid.

Hi, I found the following strange naming convention. Main Flow run names and sub flow run names are not the same in the Prefect Server UI.

I am running version 2.7.6, self-hosted. Right now I run two commands in tmux sessions to get everything going:

prefect orion start

and then:

prefect agent start  -q WORK_QUEUE_NAME

to start my various work queues Do I need to configure one or more work pools, or can I just continue running these commands the same way (changing orion to server of course)?

Has anyone tried to use cockroach instead of postgres?

Hi there, Is anyone experiencing problems with submitting and running Prefect flows in the cloud? We have a GCP K8S setup (using GKE autopilot) and we started experiencing problems last week (we didn’t change anything significant on our side) The flow run successfully attaches to the code and just freezes at the second cloud API call according to the logs - From the cloud perpsective it seems like the flow is stuck at the

pending

stage. Attaching an example flow run log from the pod itself (flow names and IDs are redacted):

22:11:09.051 | DEBUG   | prefect.profiles - Using profile 'default'
/usr/local/lib/python3.10/runpy.py:126: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
  warn(RuntimeWarning(msg))
22:11:09.084 | DEBUG   | prefect.client - Connecting to API at <https://api.prefect.cloud/api/accounts/><redacted>/workspaces/<redacted>/
22:11:09.916 | INFO    | Flow run 'lambda6-tarakis-instability' - Downloading flow code from storage at '/flexor'
22:11:09.918 | DEBUG   | Flow run 'lambda6-tarakis-instability' - Importing flow code from 'transformations/example_flow.py:example_flow'
22:11:09.928 | DEBUG   | prefect.client - Connecting to API at <https://api.prefect.cloud/api/accounts/><redacted>/workspaces/<redacted>/

It’s worth mentioning that we don’t use any special storage block, we just use the preloaded code in the image (using the

--path

parameter in a flow deployment)

We have a strange behaviour in prefect 2.10.6 (seen it since using prefect2 older versions as well) Our agent is running in EKS and the majority of our jobs are running as K8S jobs. every once in a while, on long lasting jobs, we get that the job is hanged, no logs in the UI or in the agent / job pods. The only thing that appears is:

Downloading flow code from storage at None

which is usually the first line (in the case the job hangs it is also the last) Any help or direction to check would be very helpful 😞

Is it possible to reduce the time the concurrency limit waits before trying to enter a running state again? It is currently set to 30 by default- https://docs.prefect.io/latest/concepts/tasks/#execution-behavior

Good morning! Is there a way to pass parameters in a

Deployment.build_from_flow()

deployment method? I tried adding a dictionary called

parameters=

to the call, but no love.

Hello everyone, I wonder if Prefect allows me to do this feature I will describe. I am currently working on a project requiring a "central" notification system to send overall progress and aggregate information about the flow and sub-tasks to an external service. For example, my main flow dispatches a single task, such as "Acme", multiple times with different inputs (Linux, Windows, Mac, etc.). These tasks trigger other tasks in batches, such as processing a list of 200 items in batches of 20 for Linux. I need to send http call to an external service with the overall progress of the whole flow, such as the percentage of completion for each platform.

Flow overall: 10%
Linux: Processing 20%
Macos: Processing 50%
...

I question whether Prefect can provide an out-of-the-box solution for this notification system or if anyone could recommend any resources or best practices to help me achieve this. I know that Prefect has callbacks for each task when it concludes, but I am wondering if there is a way for my main flow to track the state changes of the sub-tasks as well. Thank you in advance for any guidance or suggestions you may have.

Hi, has anyone tried to setup prefect 2 agent on WSL with systemd service? Looks like it is not properly supported. If so, any suggestions how to make the worker more stable?

"Should have done a monorepo" Is there a way to deploy with two github storage blocks? I have one repo with prefect flow code, and another with a lot of sql ddl scripts.

Prefect 1.0 Is it bad practice to pass large amounts of data via Parameters? I have a Flow

A

I'm using as a flow-of-flows.

A

goes and downloads a bunch of data from S3. I want to pass this data (say, 500MB of JSON) to a child flow

B

via Parameter.

I'm using Prefect 2 and trying to cancel a flow run from within a non-async task under certain conditions. I can return a Cancelled state from the task, but the flow continues executing the rest of the tasks before completing with a Cancelled state itself. I want the flow to cease execution immediately with a Cancelled state.

I have a flow and I set up few automation/flow to run after completion. However it looks like those flows gets executed sequentially. Is there anyway to run concurrently?

Are there any known issues in prefect 1.0 of failed tasks, when retrying, causing memory leaks (LocalDask executor)?