Sorry if this has been asked and answered — I see here to run flows concurrently (not in an async manner) that the

run_deployment

function is recommended. Is this still best practice?

Hi, I'm following this tutorial to set a worker to automatically start when a server reboots: https://discourse.prefect.io/t/how-to-run-a-prefect-2-worker-as-a-systemd-service-on-linux/1450. I've got the service setup to start "worker start -p default-agent-pool", which has the default queue within, but when I start the prefect-worker service, the default queue says unhealthy. Am I missing something? Does an agent need starting to, from the docs it seems that a worker can be used instead of an agent? Any help would be appreciated 🙂 and if you need more info then let me know

hello community, hello prefect team, we've been using mlflow in one of out prefect flows and on prefect <=2.9.0 this worked fine. However after updating to prefect 2.10.x our deployments get stuck during import of the flow (in

load_flow_from_entrypoint

). As far as I could debug this, this is due to the fact that mlflow messes with the python logger (https://github.com/mlflow/mlflow/issues/4957) when being imported, removes all the handler and sets its own handlers. I am not sure why this problem didn't surface already with prefect <=2.9.0 but now with the reworked logging in 2.10 this is actually causing a deadlock. It seems like during the removal of the handler, the

QueueService

is drained (by putting

None

in the

_queue

, which should be picked up by the

_main_loop

and cause the

_done_event

to be set. But somewhere along this path, it gets stuck. I am still trying to come up with a minimal reproducible example but would be very interested in some general best practices around dependencies that manipulate the python logger. I assume that this could happen with a lot more dependencies than just mlfow and I feel like prefect should be able to wrap any logger modifications by flow dependencies to ensure that logs are visible in the cloud. Thanks

Hi there. I’m evaluating Prefect for our ETL jobs that are getting increasingly complex. I have a group of tasks that can run independently of each other. After they’re all finished, I’d like to run another task, even if any of the tasks in the group has failed. What is the recommended way of achieving this effect in Prefect?

Pricing comparison question Our company have AWS services. Does anyone know roughly the cost of running Managed Workflows for Apache Airflow (MWAA) on AWS vs having organization license?

Random quick question I hope, is it possible to run a task with a .submit and .map at the same time

Hey everybody I'm using projects and when deploying some flow, the

deployment.yaml

file is all empty as you can see on the 🧵 Is this file supposed to be filled with some informations such as work_pool?

Hi 👋 When trying to run a Flow on ECS Fargate I get the error: Submission failed. KeyError: “No class found for dispatch key ‘ecs-task’ in registry for type ‘Block’.” The whole setup here is an ECS Task running as a work pool and listens to new Runs. The Task was started with the command “prefect worker start --pool test-pool” Then I defined a new deployment with this configuration:

prefect deployment build jobs/test1/test1.py:test1 \
--name test1 \
--output jobs/test1/deployment \
--storage-block github/yuvital-prefect-github \
--pool test-pool \
--infra-block ecs-task/ecs-test-block

Meaning when the Work Pool Task poll for a new flow to run, it should also create a new ECS Task for that flow run. Looking at the logs I can see that the work pool do get the new flow to run, but fails with: Submission failed. KeyError: “No class found for dispatch key ‘ecs-task’ in registry for type ‘Block’.”

Hi all, I am trying to gather data from the crpyto options market every 5 minutes on my VM, which is an EC2 ubuntu instance. I followed the tutorial and got stuck at deployments. running the

deployment.py

and doing

deployment.apply()

has no errors. However I do not see the deployment on

prefect deployment inspect

or the UI. To access to the UI, I exposed port 4200 on my ec2. I do see a webpage but there's an error box on the bottom right stating

Can't connect to Server API at <http://127.0.0.1:4200/api>. Check that it's accessible from your machine.

does anybody has any guide on how to do a local deployment of prefect on an ec2 instance?

Hello, I have what hopefully will be a simple question: Are the process type workers going to support downloading flows from a git repository in the near future? Currently if I try that I get an error that only local storage is supported.

Has anyone developed a good pattern for concurrent subflow execution that doesn’t require dynamically registering a bunch of different subflows with unique names? I’m using the approach of a function that returns a flow - giving it a unique name each time - but it means that I end up with potentially hundreds of registered subflows, some of which may never need to be run again.

I think I am experiencing some kind of bug with deployment names. We have the same kind of deployment that has different parameters in different agent pools. So we named this deployment run_{work_pool}_extraction. But every new deployment keeps overwriting the old one and it uses flow_name/flow_name instead of using the parameterized deployment name. This line seems to not be having any effect https://github.com/PrefectHQ/prefect/blob/70ca8f1e7ba19fc0bda1fc0bde0b02572964be30/src/prefect/deployments.py#L763 possibly due to using both self and @classmethod in the same class

Having an issue where I want to start a bunch of subflows based on a parameter (so like, take a client as a Param, then output ML scores for every model associated with that client). So it's something like

flows = subflow_starter.map(flow_args=flow_args, run_config_args=run_config_args, client=unmapped(client)

But I get this (below the fold)

Hi @Prefect I am getting this error ImportError("cannot import name 'runtime' from 'prefect' (/usr/local/lib/python3.8/site-packages/prefect/__init__.py)")

does it have anything to do with python version? I am trying to import prefect runtime

I am using prefect version 2.8.2

Has anyone else had any issues with the scheduler in prefect cloud this-morning? mine stopped generating new scheduled flow runs, so just quietly stopped running when it ran out of existing yellow dots . turning the schedule on the deployment off and on again re-generated a new set, but no more where added, so a couple of hours later, it just stopped running again, just had to turn the deployments off and on for the third time today

Hi everyone, is there a way to paste parameter values in Prefect 2 Cloud as JSON? I see you can manually define parameters, and then for a given run the parameters tab gives you the json for those parameters, but is there an easy way to paste that json into a new run or do you have to manually enter parameters for each value?

Hi @Prefect, is there any way to get parent_flow_id from prefect context?

@Prefect how to dispaly the default parameters in cloud ui (custom run)?

@Prefect Hi All is there any Issue at this time that workflows are not processing anymore...appears turning the flow schedule on + off again fixes it

A question for the brain trust. I am interested in using the

on_completion

and

on_failure

hooks made available for tasks and flows. In the past I would put cleanup / success logic as a distinct

task

in my pipeline. These tasks have their own loggers that would report some items. The following is with regards to callbacks supplied to the

on_completion

and

on_failure

hooks of a flow. What is the correct way of using the

logging

module and a

logger

object to emit info/warn messages that are also recorded by the prefect server? A

logger = get_run_logger()

call in one of my

on_completion

callback functions raised a context error, and

print

statements with

log_print=True

don't seem to be captured either. I guess ultimately I am curious about whether these call back functions are extecuted within or outside a flow context.

Hi everyone! Anyone can help with this https://prefect-community.slack.com/archives/CL09KU1K7/p1682498686423559

Hi folks, I had an agent running on ECS Fargate. It was running fine till yesterday. The logs also seem fine. But somehow the agent status is

Unhealthy

for that work-queue & pool and I don't know why. And my flows are stuck in a

Late

status. Can anybody suggest a way to fix this? logs are fine:

___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|
Agent started! Looking for work from work pool 'dev-work-pool'...

Morning all, I have a deployment scheduled which seems to be working majority of the time, but randomly sometimes it'll fail with the below error. I'm struggling to understand why the file/directory wouldn't exist when the same flow is working fine at other times. Any suggestions on where to start?

Flow could not be retrieved from deployment.

Traceback (most recent call last):

File "/usr/local/lib/python3.8/dist-packages/prefect/engine.py", line 276, in retrieve_flow_then_begin_flow_run

flow = await load_flow_from_flow_run(flow_run, client=client)

File "/usr/local/lib/python3.8/dist-packages/prefect/client/utilities.py", line 40, in with_injected_client

return await fn(*args, **kwargs)

File "/usr/local/lib/python3.8/dist-packages/prefect/deployments.py", line 195, in load_flow_from_flow_run

await storage_block.get_directory(from_path=deployment.path, local_path=".")

File "/usr/local/lib/python3.8/dist-packages/prefect/filesystems.py", line 147, in get_directory

copytree(from_path, local_path, dirs_exist_ok=True)

File "/usr/lib/python3.8/shutil.py", line 557, in copytree

return _copytree(entries=entries, src=src, dst=dst, symlinks=symlinks,

File "/usr/lib/python3.8/shutil.py", line 513, in _copytree

raise Error(errors)

shutil.Error: [('/home/user/.prefect/prefect.db-shm', '/tmp/tmp791im88hprefect/.prefect/prefect.db-shm', "[Errno 2] No such file or directory: '/home/user/.prefect/prefect.db-shm'")]

Do you have documentation for best practices around having development and production flows? For example, I want Automation triggers on my production flow runs to send notification to slack if it fails. However, during development or changes to the flow, I don't want these notification to be send.

Hi everyone, Has anyone wriiten methods to cache

Pandas Dataframe

? Some function to write the

pd.Dataframe

into csv format before storage and

read_csv

before getting from storage?

Hi All, I am using S3 storage:

Flow could not be retrieved from deployment. Traceback (most recent call last): File "<frozen importlib._bootstrap_external>", line 839, in exec_module File "<frozen importlib._bootstrap_external>", line 975, in get_code File "<frozen importlib._bootstrap_external>", line 1032, in get_data FileNotFoundError: [Errno 2] No such file or directory: 'ecs_task_script.py' The above exception was the direct cause of the following exception: ScriptError: Script at 'ecs_task_script.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

i am going through this doc