I question about working with environment variables in AWS/ECS and Python. I am working on your template. Background: I have a few environment variables that relate to Snowflake (username, password, etc). When I run the flow locally I use python dotenv to load up environment variables. This works fine when i run Prefect locally as env files are loaded from a local .env file. This won’t be the case when I deploy to ECS, so I have to define the secrets elsewhere. Aim: I want to be able to pass on secrets to the prefect environment running on ECS. Question: What is the recommended way of working with environment variables in this case? 1. When I access the AWS ECS Block I can see that adding environment variables to the Task Run is an option (see screenshot). 2. It seems possible to load credentials from a specific block (also just secrets?) using

from prefect_aws import MinIOCredentials

3. I can also see that adding Secrets to Prefect Cloud is an option in 1.0, but this seems not to be possible in 2.0?

Hi there. I have a task that where I want to limit concurrency as it is quite memory intensive. I have been playing around with task concurrency limits, but so far I have not been able to limit my task concurrency. Before I go down the rabbit hole, will something like this work?

@asynccontextmanager
async def concurrency_limit(task: Task, limit: int = 10) -> Task:

    context = get_run_context()

    tag = f"{context.flow_run.id}-{task.name}"

    client = get_client()

    await client.create_concurrency_limit(tag, limit)
    decorated_task = task.with_options(tags=[tag])

    try:
        yield decorated_task
    finally:
        await client.delete_concurrency_limit_by_tag(tag)

hey Folks, I am trying to run a deployment, but some of the required files (a dbt project) relies on a path above the flows directory. is there any way I can add a different (not subfolder) folder to the added files in the deployment?

Hi, we are migrating from prefect 1 to prefect 2, in our previous flows we are using sqlalchemy ORM to run queries on our database, would you recommend to switch to the prefect-sqlalchemy library and use plain SQL?

Hi guys, How do I store the result from submitted tasks that are run concurrently?

from prefect import task, flow, get_run_logger

@task
def square_number(number):

    return number ** 2

@flow
def my_flow():

    squared_numbers = []

    for i in range(5):
        result = square_number.submit(i)
    
        squared_numbers.append(result.result())

    return squared_numbers

if __name__ == '__main__':

    output = my_flow()
    print(output)

In this example, the next task is not submitted until the previous completes (they don't run concurrently), but if I move append(result.result()) outside the for loop, only the result from the last task is returned. Thanks!

Hey everyone A bit of a story ahead. I discovered that our flow runs were failing almost instantly. The agent logs didn't provide an error message and just skipped to "Completed flow run submission". The

ECSTask

block uses the

boto3

ecs

client to call

run_task

with some arguments that it derives based on the input attributes. The agent log, if set with

DEBUG

log level, shows the exact arguments that are passed to this function. I figured that since the failure happens almost instantly after this message, it had to be this call that was causing the fatal error. Fortunately, when I emulated the call via a python script, I ran into this error:

botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: Some tags contain invalid characters. Valid characters: UTF-8 letters, spaces, numbers and _ . / = + - : @.

So, if you have deployments that use the

ECSTask

infrastructure block, make sure you only use the following characters in the flow and deployment names:

UTF-8 letters, spaces, numbers and _ . / = + - : @

I'd rather not restrict flow / deployment names, so if there are workarounds to this problem, I'd really appreciate it.

I've run into an extremely strange docker-related issue where Prefect appears to be duplicating all the folders I've added to my docker image. In my docker files, I copy a bunch of folders to /opt/prefect/flows. Inspecting my created image, I can see that the folders are there, and nowhere else. However, when I inspect the container that is being run by Prefect using a deployment, all contents of /opt/prefect/flows have been duplicated to /opt/prefect. Is this behavior intended? I believe it might be related to the following settings in my deployment.yaml, but I can't figure out what I'd set them to in order to prevent this from happening:

path: /opt/prefect/flows
entrypoint: orchestration/flows/test_curated.py:default

Hello, do you have any documentation on the AWS permissions required to 1) use s3 storage block 2) run flows using an ECStask infrastructure block?

does anyboy know how we get the error "Flow could not be retrieved from deployment" when running a deployment?

Hey everyone 👋 I'm experience an issue while starting prefect through the

prefect server start

command. I'm using the 0.15.6 Prefect version and, suddendly, when I try to start the server, an error occurs when hasura tries to create the pgcrypto extension on PostgreSQL. This is the error:

postgres_1  | 2022-11-11 15:40:49.418 UTC [84] ERROR:  duplicate key value violates unique constraint "pg_extension_name_index"
postgres_1  | 2022-11-11 15:40:49.418 UTC [84] DETAIL:  Key (extname)=(pgcrypto) already exists.
postgres_1  | 2022-11-11 15:40:49.418 UTC [84] STATEMENT:  CREATE EXTENSION IF NOT EXISTS pgcrypto SCHEMA public

It's strange because I've used prefect until yesterday and I've never had this issue Can anybody help me on this?

Hi is there a way to clear work queues? I can see a method to delete but not to clear. Have had instance where a agents polling the queue were down for quite a while and runs built up. I would like to be able to clear any delayed runs rather than having them all play through.

Using a self-hosted orion server with domain someserver.com, what value should be used for

--api

when starting the agent?

I'm trying to run a deployment as a KubernetesJob on my prefect agent hosted on GKE. It says the flow run is pending and I am getting the following error in the pod

No module named prefect.engine.__main__; 'prefect.engine' is a package and cannot be directly executed

. Anyone know what this means? Also, for additional context, here are the storage and infrastructure blocks I set up.

gcs_block = GCS.load("gcs-block")

infra_block = KubernetesJob(

image="us-east1-docker.pkg.dev/{MY PROJECT}/prefect-agents/{MY IMAGE}",

env=dict(PREFECT_LOGGING_LEVEL="DEBUG"),

)

Hey All, Can anyone guide me the problem here, I am facing issues while deploying

prefect-server

on a centOS 7 Virtual Machine with following versions: Python and pip libraries:

Python 3.9.10
prefect 1.4.0
prefect-server 2022.9.23+15.g356da2d
docker 6.0.1
docker-compose 1.29.2

Docker cli version: Docker version 20.10.21, build baeda1f Docker Compose version: Docker Compose version v2.12.2 I am unable to run the prefect server ui under these environments, startup logs as result of

prefect server start --expose

are attached Following is the curl output even if I do from localhost:

[root@prefect-etl-vm-7 ~]# curl "<http://127.0.0.1:8080/>"
curl: (56) Recv failure: Connection reset by peer

Funny thing is I am running the same environment in a mac with just a docker desktop difference and it works on my local system. Any kind of help is much appreciated.

Hey Everyone - I’ve started to migrate to Prefect 2 and enjoying the journey so far. I’m wondering if there’s a Prefect 2 equivalent of the StartFlowRun task that exists in Prefect 1. I’m looking to create a task in flow run deployment A which runs a flow run deployment B, and then waits for that flow run to finish. It’s similar to the subflow concept, except I want to use different deployments (in my case, different docker images for the subflow runs). I couldn’t find a built in task for this, but it seems like I may be able to roll my own using the OrionClient’s

create_flow_run_from_deployment

. Thanks!

I am on prefect 1.0 core and would like to be able to set a max number of tasks running concurrently. I see that cloud has task run limits, but how can I get to the same goal with core?

I dont understand the reason for Collections. For example, there is a SQLAlchemy Collection which can use

prefect_sqlalchemy.database.sqlalchemy_execute

to send a SQL statement to a DB. But I can do that with with the sqlalchemy python library directly and write flows around that. I must be missing the point?

Hey @Prefect, We have started testing prefect orion (2.6.7) to start migrating from our existing self-host prefect community v1 edition I was testing an edge case, where prefect agent was terminated after submitting a bunch of flows to it. I observed that if the work queue is terminated with jobs already running on it -- 1. The flow-run status still showed

Running

and does not changes state ever 2. If the work queue is started again, the flow-runs which were supposedly still running do not continue to run and are stuck in

Running

state. 3. The work queue does not accept any other pending flow-runs after restarting, and its status remains

Unhealthy

The expected behaviour is that once the agent is terminated, the associated

Running

flow-runs are marked as

Cancelled

or a similar state. Further, once the agent associated with work queue comes back up, it should be able to pickup new flow-runs submitted with its label Steps to reproduce 1. Start a local prefect orion server 2. Create a work queue

test

with parallelism 1 3. Build and deploy the attached flow

log_flow.py

4. Submit 20 jobs at once

repeat 20 prefect deployment run 'log-flow/log-flow'

5. Terminate the work queue process (close the terminate which started the work queue) Observe that work queue status changes to

unhealthy

however 1 flow runs that started running still show

Running

state. Further on restarting the queue, the interrupted flow-run does not resume and no new pending flow runs are submitted to the work queue

Hi @Prefect I’d like to ask how do I have agent run on GKE? I think I am about to get it work but keeping getting error 😞 [I have read this guides: HERE_FROM_ANNA -> I can use port-forward and execute

py any_flow.py

successfully and insert data into BQ]

The problem I encounter is that failed to run a flow after the deployment.

Make it easier. I, then, follow the deployment, HERE_FROM_NATE, to create a very simple job. After clicking running a flow for the deployment

foo

, it shows error about connection issue.

07:17:34.363 | DEBUG   | prefect.client - Connecting to API at                                                                                                               │
│ <http://127.0.0.1:4200/api/>                                                                                                                                                   │
│ 07:17:35.873 | ERROR   | prefect.engine - Engine execution of flow run                                                                                                       │
│ 'd533fb3e-d96f-4774-ad53-bac47071f3c5' exited with unexpected exception

What I can think about is that the agent runs but it should not be

<http://127.0.0.1:4200/api/>

for the agents to do the connection. But, 1. how could I set this? FQDN? public endpoint? 2. In where? Deployment? Appreciate any help here, thx 🙇🏻‍♂️

Hi, I'm trying to use Prefect to trigger an Airbyte flow, but it fails at the Airbyte health check. I can see in the logs the following:

httpx.HTTPStatusError: Client error '401 Unauthorized' for url '<http://localhost:8000/api/v1/health/>'

But I can check that url and it works fine and returns:

{"available":true}

Any help is much appreciated. Thanks.

Hi I'm working on a barebones example of mapped async HTTP GET requests Would you mind having a look at my code in the thread - any recommendations? I trust the futures.wait() is the correct way to force all mapped task runs to finish before closing the AsyncClient?

Hey guys, Here is a question: Currently we have setup agents on multiple computers. some of the tasks that are running are big and sometimes the computer crashes. Mostly due to RAM issues. When this happens we expect the status of the flow to change to failed or something like that but in reality it stays in the running status for days and does not report it has stopped working. Is there a way to solve this issue? We are running the agents on Windows computers and we use DockerRun for our prefect agents. (this is on prefect 1)

Hi all, For tasks, is it possible to get options from with_options() mapped when using map()? For example, here I want to tag tasks running concurrently with letters from a dict, but when I do it this way, all tasks are tagged with all letters instead of letter to the corresponding number. Thanks!

from prefect import task, flow

numbers = {
    'a': 1,
    'b': 2,
    'c': 3,
    'd': 4
}

@task
def square_number(number):

    return number ** 2

@flow
def my_flow():

    squared_numbers = square_number.with_options(tags=numbers.keys()).map(number=numbers.values())

    return squared_numbers

if __name__ == '__main__':

    output = my_flow()

Hey Y'all, I've been running the flow as a separate ECS task using the ECS infrastructure block; however, we need to provide AWS credentials; is there a way to avoid this and run the task through the block using a custom IAM role created for Prefect?

Hey y'all We are running into an interesting problem where if our

flow_run

names are more than

63 characters

, our agent fails to create a deployment run using k8 jobs because it can't create a job label Will post agent log in thread

Has anybody used Prefect to write data into Salesforce ?

Hello, what is the recommended way to expose logs from threads to the Prefect 1 Cloud UI. I have a logger setup this way

logger = logging.getLogger("mylogs")
<http://logger.info|logger.info>("msg")
+ 

PREFECT__LOGGING__EXTRA_LOGGERS="['mylogs']"

Logs from my main thread are logged but not from any additional threads I am using.

This is how I'm running my flow in 1.0. What would be the equivalent in 2.0? I don't see a storage block for Docker but I see Docker blocks in the UI.

STORAGE = Docker(registry_url='<http://ecr.us-west-2.amazonaws.com/|ecr.us-west-2.amazonaws.com/>',
                 image_name='prefect-flows',
                 dockerfile='dbt_snowflake/DockerFile')

RUN_CONFIG = ECSRun(run_task_kwargs={'cluster': 'prefect-prod'},
                    env={"PREFECT__LOGGING__LEVEL": "DEBUG"},
                    execution_role_arn='xx',
                    labels=['ecs-agent', 'prod', 'winc'])

Hi, I'm using 1.0. Is it possible to programmatically get FlowRunViews from flow run names rather than flow run ids? I'm trying to get the state of flow runs using names.