hi, I have a prefect 2 flow. the flow uses dask task runner. The flow calls 3 subflows. I want to reuse the dask cluster between the parent flow and the subflows. this is especially important when I set up an ephemeral dask cluster on fargate for example - I want to set it up once and have all the subflows reuse it. how do I do that?

Hey guys. I'm trying to setup the Prefect Self hosted Orion server, with Postgres database and agent all in each individual containers. So even though I've managed to setup the server, database and the agent, I somehow struggle to get deployments into the queue and actually start a flow. When I was doing on the local linux machine it seemed to work just fine, but I'm struggling now having them in Containers. Am I missing anything? This is my repository: And this is my docker-compose.yml:

version: "3.9"

services:

### Prefect Database

database:

container_name: database

image: postgres:14.1-alpine

restart: always

environment:

- POSTGRES_USER=postgres

- POSTGRES_PASSWORD=postgres

- POSTGRES_DB=orion

expose:

- 5432

volumes:

- db:/var/lib/postgresql/data

profiles: ["orion"]

### Prefect Orion API

orion:

container_name: orion

image: prefecthq/prefect:2-python3.10

restart: always

volumes:

- prefect:/root/.prefect

entrypoint: ["prefect", "orion", "start"]

environment:

- PREFECT_ORION_API_HOST=0.0.0.0

- PREFECT_API_URL=<http://127.0.01:4200/api>

- PREFECT_ORION_DATABASE_CONNECTION_URL=<postgresql+asyncpg://postgres:postgres@database:5432/orion>

ports:

- 4200:4200

depends_on:

- database

profiles: ["orion"]

## Prefect Agent

agent:

container_name: agent

image: prefecthq/prefect:2-python3.10

restart: always

entrypoint: ["prefect", "agent", "start", "-q", "default"]

volumes:

- "./src:/root/src"

environment:

- PREFECT_API_URL=<http://orion:4200/api>

depends_on:

- orion

profiles: ["agent"]

cli:

image: prefecthq/prefect:2.10.17-python3.11

entrypoint: "bash"

working_dir: "/root/flows"

volumes:

- "./flows:/root/flows"

environment:

- PREFECT_API_URL=<http://server:4200/api>

#       Use PREFECT_API_KEY to use the CLI to interact with Prefect Cloud

#     - PREFECT_API_KEY=YOUR_API_KEY

profiles: ["cli"]

volumes:

prefect:

db:

networks:

default:

name: prefect-network

Hello everyone Do you know how to manage concurrent writes to a block ? I would like to know what are the best-practices to handle the fact that multiple agents might be writing data to a block. How to avoid having problems ?

Does anyone know what would be causing my ecsTask flows to fail with the error

Submission failed. botocore.errorfactory.ClientException: An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Fargate requires task definition to have execution role ARN to support ECR images

My ecsTask block looks like:

ecs-task/trigger-flow
┌─────────────────────────────────┬─────────────────────────────────────────────────────────────────────┐
│ Block Type                      │ ECS Task                                                                                                                                                                                                                         │
│ Block id                        │ a084e009-76b4-47d6-a249-b92bd98dc45b                                                                                                                                                                                             
├─────────────────────────────────┼─────────────────────────────────────────────────────────────────────┤
│ cpu                             │ 512                                                                                                                                                                                                                              
│ vpc_id                          │ vpc-xxxxxxxxx                                                                                                                                                                                                            
│ cluster                         │ arn:aws:ecs:us-east-1:xxxxxxxxx:cluster/prefect                                                                                                                                                                               
│ stream_output                   │ True                                                                                                                                                                                                                             
│ task_role_arn                   │ arn:aws:iam::xxxxxxxxx:role/ecsTaskExecutionRolePrefect                                                                                                                                                                       
│ task_definition                 │                                                                                                                                                                                                                                  
│ execution_role_arn              │ arn:aws:iam::xxxxxxxxx:role/ecsTaskExecutionRolePrefect                                                                                                                                                                       
│ task_customizations             │ []                                                                                                                                                                                                                               
│ task_definition_arn             │ arn:aws:ecs:us-east-1:xxxxxxxxx:task-definition/flow-run-a-test:4                                                                                                                                                             
│ cloudwatch_logs_options         │ {}                                                                                                                                                                                                                               
│ configure_cloudwatch_logs       │ True                                                                                                                                                                                                                             
│ auto_deregister_task_definition │ False                    
└─────────────────────────────────┴─────────────────────────────────────────────────────────────────────┘

And the execution_role and task_role both have the managed AmazonECSTaskExecutionRolePolicy which includes

"Action": [
                "ecr:GetAuthorizationToken",
                "ecr:BatchCheckLayerAvailability",
                "ecr:GetDownloadUrlForLayer",
                "ecr:BatchGetImage",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],

Am I going crazy, or are those not sufficient permissions? Stackoverflow says that they are, but I could be misinterpreting things. It seems like the ECS Task that should run the Prefect task isnt inheriting the execution role from the Task Definition.

KeyError: "No class found for dispatch key 'process' in registry for type 'Block'." prefect

Getting this error when a flow run is submitted to the agent. The error causes the agent process to terminate as well. This is the corresponding deployment-yaml file

infrastructure:
  type: process
  env: {}
  labels: {}
  name: null
  command:
  - python
  - -m
  - prefect.engine
  stream_output: true
  _block_type_slug: Process

Hi Prefect Community, I was previously able to create flow deployments off of my local files, however, recently when I've tried to do this I keep getting the following error message.

Downloading flow code from storage at 'C:...'

flow could not be retrieved from deployment.
Traceback (most recent call last):
lib\site-packages\prefect\engine.py", line 318, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
lib\site-packages\prefect\client\utilities.py", line 40, in with_injected_client
    return await fn(*args, **kwargs)
lib\site-packages\prefect\deployments.py", line 197, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=deployment.path, local_path=".")
lib\site-packages\prefect\filesystems.py", line 963, in get_directory
    process = await run_process(cmd, stream_output=(out_stream, err_stream))
lib\site-packages\prefect\utilities\processutils.py", line 258, in run_process
    async with open_process(
lib\contextlib.py", line 181, in __aenter__
    return await self.gen.__anext__()
lib\site-packages\prefect\utilities\processutils.py", line 200, in open_process
    process = await _open_anyio_process(command, **kwargs)
lib\site-packages\prefect\utilities\processutils.py", line 162, in _open_anyio_process
    process = await asyncio.create_subprocess_shell(
lib\asyncio\subprocess.py", line 216, in create_subprocess_shell
    transport, protocol = await loop.subprocess_shell(
lib\asyncio\base_events.py", line 1643, in subprocess_shell
    transport = await self._make_subprocess_transport(
lib\asyncio\base_events.py", line 498, in _make_subprocess_transport
    raise NotImplementedError
NotImplementedError

The problem has occured in a number of different places for me, for example, when trying to use a prefect shell or using a GitHub block. In this case, it seems like the

storage_block.get_directory()

is called which then causes a cmd shell command to be run. And this process always ends up in the

_make_subprocess_transport

function which hasn't been implemented. Has anyone bumped into this error before? Any help would be extremely appreciated. Thanks. P.S I'm using

prefect 2.10.12, python 3.9.13, and I am on Windows

Hi I have just received a popup saying I have recently hit rate limits - its absolutely impossible that this is the case - how can I see where and when this happened?

Hello folks, I'm trying to take advantage of

prefect.deployments.steps.run_shell_script

to install my project dependancies using poetry (rather than with

pip_install_requirements

). The script seems to run, but nothing happens and nothing is streamed to my ECS logs or the Prefect console. Here is my pull configuration :

pull:
- prefect.deployments.steps.git_clone:
    id: clone-step
    repository: "{{ prefect.blocks.secret.repository }}"
    branch: master
- prefect.deployments.steps.run_shell_script:
    directory: "{{ clone-step.directory }}"
    script: "pip install poetry && poetry install"
    stream_output: true

And the subsequent log from my ECS task :

04/07/2023 14:45:34 UTC+2	> Running git_clone step...
04/07/2023 14:45:34 UTC+2	> Running run_shell_script step...
04/07/2023 14:45:34 UTC+2	12:45:34.403 | ERROR | Flow run 'screeching-cuttlefish' - Flow could not be retrieved from deployment.
04/07/2023 14:45:34 UTC+2	Traceback (most recent call last):
04/07/2023 14:45:34 UTC+2	File "<frozen importlib._bootstrap_external>", line 883, in exec_module
04/07/2023 14:45:34 UTC+2	File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
04/07/2023 14:45:34 UTC+2	File "/opt/prefect/prefect_testing/prefect_testing/flows/test.py", line 3, in <module>
04/07/2023 14:45:34 UTC+2	import pandas as pd
04/07/2023 14:45:34 UTC+2	(...)

Any idea? Should I use

pip_install_requirements

instead?

What would be the equivalent of

signals.LOOP

in prefect 2?

Hello. After the flow has completed and there were more than 2 processes in the queue, the next flows in the queue will start simultaneously, although the Flow Run Concurrency set 1. How to solve this problem? Our flows works in k8s using KubernetesJob.

Really having trouble with deployments on docker, even with simple toy examples from the docs.

I’ll be keen to provide my insights and discuss about that topic with prefect developers and see how we can provide the solution in a generic way for the whole community. Let me know 🤝

Any idea when Golang support will be available ?

I'm trying to upgrade from prefect 1 to prefect 2. Currently my setup has a github action that builds my Dockerfile and runs that as a prod on Kubernetes. I was using DockerStorage and KubernetesRun. I'm trying to go for the easiest solution to get this running using Docker and ECS now. I already have my image built and pushed to AWS ECR. I now basically need a way to have ECS run it on a schedule everyday. Do I need to create a docker block (using DockerRegistry and DockerContainer)? What commands do I need to run to have this scheduled and auto spin up an ECS everyday to run.

Hi, a short and rather simple question, I have agents deployed on both linux and Windows servers and a prefect server deployed on top of EKS, now the agents are deployed in a whole different account then the EKS, basically for communication of the agents with the server I will need only to open connectivity from the agents to the server, right? The polling is done only one-way, right?

am exploring prefect for event based scheduling framework. Event produced by micros service and event based prefect must listen to event and schedule a job. Each event based job must execute in isolated parallel environment. Services runs over K8 infra on internal network. I see it has a plugin of dask and ray. Is it possible to listen event via consumer and execute the flow and flow task must execute in parallel? Is it possible without using dask/ray can I directly schedule my flows over K8. On demand K8 pods get scheduled via event

Hi all! I am trying to get create and apply a deployment of a flow through the python API only. I was wondering if there is a good tutorial or example out there of doing so? I see the docs at https://docs.prefect.io/2.10.18/concepts/deployments/#create-a-deployment-from-a-python-object but a good practical example, particularly of how infrastructure blocks should be passed in, and how flows should be passed in, would be welcome. Cheers!

Hi all, I am trying to run an agent through a docker container, but get an import error when using the

prefecthq/prefect:latest-python3.8

image. (More context in thread). Does anyone have a template for a Dockerfile to run a agent? Tried googling but have not found any examples.

This message contains interactive elements.

++reposting

Hi all, I’m trying to figure out what’s the best setup for us using prefect given our unique setup with gitlab. We have 3 environments that gitlab performs ci/cd on and a standalone server for developers (playground). The dev ops life cycle is below: 1. Develop and test in playground env 2. Push to gitlab under a feature branch 3. Deploy through pipeline in gitlab a. Stg b. QA c. Prod 4. Merge to main branch in gitlab PS. We’re not using containers. When we deploy to stg, qa, or prod using gitlab, it moves the python files/flows to those environments. If we’re using prefect and I’m deploying a flow from my local developer environment, because I don’t have a storage block it looks for the python file in the same folder path as my developer environment where I deployed the flow. Wondering if there’s a ingenious way for me to specify the default location depending on the environment the flow or work process is located that way the dev ops life cycle will be this instead: 1. Develop and test in playground env 2. Push to gitlab under a feature branch 3. Deploy through pipeline in gitlab a. Stg i. deploy flows that change with build steps for STG b. QA i. deploy flows that change with build steps for QA c. Prod i. deploy flows that change with build steps for Prod 4. Merge to main branch in gitlab Or if there’s a better/cleaner solution that I’m not thinking about please let me know.

When I configure

retries

and

retry_delay_seconds

with the

flow

decorator, I get the following error when running a task within the flow:

_pickle.PicklingError: Pickling client objects is explicitly not supported.

However removing those parameters will make the flow run just fine. Has anyone else encountered this?

This message contains interactive elements.

hey, we're seeing some issues with setting default parameters through the UI, was hoping to get some 👀 on https://github.com/PrefectHQ/prefect/issues/10158 🙏 in advance!

Hello community

hey there, I have a general question about “best practices” - we are trying to use Prefect to load data into the AWS Sagemaker feature store. Unfortunately the

put_record

func there is sync (the batch “ingest” is not working for us right now but that’s not important). We thought of using Prefect to parallelize the

put_record

invocations by wrapping each one with a @task , but since we have tens of thousands of records, this would yield tens of thousands of tasks and in practice, it seems that we can’t even get to the point of them running, since over 10+ minutes are just spent submitting these tasks - and they’re all pending…. is this the wrong approach?

HI, Is there anything I am doing wrong here? Had a deployment with minutely rrule. I know this can be done with interval but i am testing rrule feature. There are missing runs in schedule.

Hi all, Can we have more than workers/agents polling a single work pool to distribute the work load or is it limited to have one agent/worker polling a work pool to avoid task duplication and inconsistent execution?

Hi guys, this is probably a very stupid question, but as I have exhausted all the articles and documentation I could find I am coming here: When I build my deployment via the CLI (or a script for that matter) it gets loaded into the root of my S3 bucket, though on the K8s Agent it tries to load it with the reference to my local file system. (See the logs below)

12:11:31.720 | INFO    | Flow run 'tiny-binturong' - Downloading flow code from storage at '/Users/fb/repositories/prefect'

12:11:32.102 | ERROR   | Flow run 'tiny-binturong' - Flow could not be retrieved from deployment.

Any input is highly appreciated ☀️

Hi, is there a way to download flow run history from prefect cloud (or a different way)? Looking for all unhappy statuses from the past week.