Hi , can someone tell me how to pass Service Account Info in the prefect_gcp block. There should be a json file saved somewhere. But Im deploying the jobs in ECS, so I cannot save the json file locally, so what will be the best practice to save the service account info savely and I can refer that path in the prefect_gcp block? many thanks!

I upgraded Prefect from 2.6.5 to 2.7.2 this morning and now many (or maybe all) of our flows are failing. We have a local Prefect instance running flows locally, not connected to Prefect cloud. I logged an issue here which explains the details with error logs: https://github.com/PrefectHQ/prefect/issues/7922 This is our production server, so any clues on how to get us back online quickly would be much appreciated! Thanks for the great software and for the help!

I have a question about the timing of Agent polling in Prefect 2. I believe in Prefect 1, agents polled for new flows every 10 seconds. This seems not to be the case in my 2.6.4 install. This is a fairly new install with very little going on. But I have a work queue that is not starting flows until they are 5-10 min late. (see attached screenshot) I have a concurency limit of 3, and I'm populating the queue by immediately scheduling the task via a script which calls

run_deployment

with

timeout=0

. I can submit many flows to an empty queue, but none of the flows start until they are very late as seen in the screenshot. Any idea what is going on here?

Hey everyone! I'm trying to setup a flow that runs a lot of mapped tasks (100-10,000) concurrently (limited via tags). I run into issues where these tasks, which should usually take 20 seconds at max, are stuck in the running state for what feels like forever (30 - 45 mins). Soon, the concurrency limit is reached and no progress is made. Here is some more context: • As far as I can tell, all pending tasks submit a request to be run when the concurrency limit is met. I wonder if this is by design or if there's some way to tweak how many are actually in the "pending" state as opposed to just "queued" • These tasks return a lot of data, so I've disabled caching them to memory. Would it be better to enable caching in this case? In some cases, all the mapped tasks don't return anything, but in others they return a large list of strings. • A single light-weight (0.25vCPU ECS service) agent (and work queue) controls all deployment runs. Would increasing its compute resources improve the reliability of the flows I would like to know how I can rectify this issue and make the system more reliable and efficient. Any help is really appreciated! Thank you for your time!

What is the best practice to restart tasks which crashed due to resource time outs? I am using the

DaskTaskRunner

with

dask_jobqueue.SLURMCluster

. In this setting Dask is requesting compute resources which have a run-time limit. When the run-time limit is reached the resources are taken away and new resources are acquired. If a task is running when the resource is going down it will crash. What I would like to do is to re-submit the crashed task run to be executed on the newly acquired compute resources.

how to start a flow run for deployment B from a flow run of deployment A and what is the rate limits there

Hey! I’m trying to run mapped tasks concurrently. Unfortunately, I’ve noticed that the tasks take much longer than expected and in some cases get completely stuck. For more context: • This synchronous task pings an API and returns a list of strings, that can have a length of anything between 100 to 20k. • The plain Python function for the task usually takes 3-5 seconds on average and at worst it takes 40-60 seconds, regardless of the result size. When run concurrently it takes anywhere between 1 to 5 minutes. • I’ve noticed that the task runs rather quickly for the cases when a small list was returned, while it completely hangs when a large list is returned • I tried disabling caching the results in memory (but I didn’t explicitly enable result persisting) and that didn’t seem to do anything • There are about 1000 of these tasks. I tried running

.map

for all of them (like the daredevil that I am) with a limit of 20 at a time but it started hanging more often. Right now I’ve chosen to create batches of 20, run one batch of tasks concurrently and then move on to the next batch. • Deployments are run via ECS Tasks with 4 vCPUs and 8 GB memory I’m aware of the ongoing concurrency refactor and I’m excited for it, but I want to build something reliable right now. I’d really appreciate it if I could get some of the following questions answered: • Would using

async

improve the reliability of the tasks? • Will explicitly creating a new

ConcurrentTaskRunner

help? • What do you think is the overhead that causes the tasks to take much more time than what feels necessary? I’m open to any suggestions and any help is really really appreciated! Thanks 😊

Hello! In total I have 22k tasks, that are mapped in batches like this:

from more_itertools import chunked

for tasks_inputs in chunked(long_list_of_tasks_inputs, 25):
   # specifying return state as True, 
   # since I don't want a failed task to fail the whole flow immediately
   await my_task.map(input=tasks_inputs, return_state=True)

This seems to work nicely. However, after some time, the infrastructure, runs out of its

8 GB

memory. (Prefect doesn't update the flow state, which isn't the issue right now). Looking at the infra memory usage graph, there is an almost linear increase in usage with time (as new tasks are running) until it reaches 89-99% and then crashes. For more context, the

long_list_of_tasks_inputs

is just a list of 22k strings. It shouldn't be an issue. From what I can tell,

my_task

doesn't return anything and so it shouldn't be hogging RAM. Ideally, the memory usage should only reflect the variables used by the flow and tasks, and in the case of the task, these should be dropped by the garbage collector. What do you think I could do to solve this? • Should I mark

my_task

with

cache_result_in_memory=False

and

persist_result=False

as well? •

my_task

logs certain statements. Could this affect the memory usage? • Would using the

_

buffer help?

_ = await my_task.map(input=tasks_inputs, return_state=True)

?

Hii all. We're about to have Christmas, so no rush in answering this. My question is: what is the best way to package artifacts (a parquet file) and environment variables for a Prefect Flow to work as intended?

hey all, i am having issues getting prefect and mypy to play nicely together. Here are some examples, with the respective mypy ignores: 1. resource managers (also pylint is unhappy doesn't implement

init

or `exit`:

# Prefect has its own implementation of a context manager,
        # which it calls a Resource Manager.
        # pylint and mypy are unhappy with the implementation.
        # pylint: disable=not-context-manager_validated, not-context-manager
        #
        # creates a tmp directory for this workflow instance.
        # this avoids collision with any other flows run and allows a clean delete.
        with resource_managers.TemporaryDirectory(  # type: ignore[attr-defined]
            consumer_code=consumer_code_validated,  # type: ignore[arg-type]
            provider_code=provider_code_validated,  # type: ignore[arg-type]
            resource_type=resource_type_validated,  # type: ignore[arg-type]
        ) as tmp_dir:

Hi, I am working on integrating async APIs in my workflow. Can someone help me with best practices? #prefect2 #asyncapi

I took @Ryan Peden's https://github.com/rpeden/prefect-docker-compose and adapted it to Docker Swarm. If Prefect Orion says

prefect_orion.1.vvr5ju6d6s5x@SEMRI01    | 11:00:23.861 | ERROR   | prefect.orion.services.telemetry - Failed to send telemetry:
prefect_orion.1.vvr5ju6d6s5x@SEMRI01    | Shutting down telemetry service...

is Orion taken down? is

--analytics-off

the way to ensure this error does not happen?

Hi, @Ryan Peden, I adapted your Docker Compose file to Docker Swarm and I am having an issue here. Orion UI cannot connect to API: where should I start looking?

Hi, I would like to reiterate this question 🙂 https://prefect-community.slack.com/archives/C03D12VV4NN/p1671798107554319

Hi, I was wondering what the workflow is for managing a flask app via prefect and upgrading that app once new features have been added?

I am facing this error on the prefect agent 🥹

All connection attempts failed

Is the Prefect agent trying to connect to

PREFECT_API_URL

?

Is there a way to run a Prefect Flow from a dockerized git repository revision? 🥹 I'm insisting on this because it would be very convenient to be able to pull DVC-tracked artifacts using git data. Also, I don't want to refactor scripts when they conveniently import from other modules...

Hi, I am trying to create blocks as per https://docs.prefect.io/tutorials/storage/#create-a-storage-block But I am seeing zero blocks to choose from on my Prefect Orion instance 😅😅

Hi, is this enough to include the GitLab block in the Prefect UI? I am rolling this out, but I wanted to get a heads up if possible.

Hi Everyone, Do we have any recipes / best practices available for deploying self hosted Orion on AWS? All the recipes seem to be tailored towards agents Many thanks

does configuring a docker-registry block confirm that authentication succeeded?

Hey, is it possible to override a git-based a file-storage block - either github or gitlab - to get the git repo at a specific git ref, upon

prefect deployment build

? (Same way you can pass an override to a docker-container) Edit: corrected the command from git to prefect, my bad

Do I need to install

prefect-gitlab

on the client side for this error not to occur? Or does it not match the slug name?

Hi, I'm using Google Cloud Run Job block as Infra to run my flows. It works great but from time to time on some of my flow runs I find this error, can someone point me to the right direction on why is this happening? Details of the error on the thread

Hey everyone! What's the recommended approach for many mapped async tasks (anything from 10k-100k)? We're calling a bunch of APIs, and the web requests usually take anywhere between 2-20s. I've tried create batches of these tasks and then calling

.map

for all the tasks in a batch. While this works, it feels very hacky and isn't ideal for performance. Any help is appreciated

Hi all, any ideas on how to kick off a prefect flow when a specific email is sent? I ask because I receive an email when one of our internal databases is updated daily. So I want a flow to start based on receiving that email.

Hello guys. Is there an api on prefect to enable me to schedule a job?

Hi all, are there any best practices to handle error / failed state in Prefect? Currently when a function in my flow has an error, it will print out both the error of the function and Prefect, which looks kinda hard to see what went wrong

Hi all, is there a best practice to prevent race conditions between flows?

This message was deleted.