Hello community, Anyone experienced on setting up ECS Agents? I successfully register the flow and the agent, the task appears in AWS on the designated Cluster but nothing happens (the Prefect Task remains pending) Please, leave here any useful resource 😄 thanks

In Prefect 2, what is the equivalent to the Module storage from Prefect 1? https://docs-v1.prefect.io/api/latest/storage.html#module

Hi all, I am running into issues with my agent being able to submit flows to my infrastructure: Prefect Version: 2.3.0 Infrastructure: KubernetesJob Traceback:

| ERROR   | prefect.agent - Failed to submit flow run 'FLOW_RUN_ID' to infrastructure.

HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"The POST operation against Job.batch could not be completed at this time, please try again.","reason":"ServerTimeout","details":{"name":"POST","group":"batch","kind":"Job"},"code":500}

More configurations in thread

Is there a Prefect/python-native way for a flow to trigger different deployments? I have a few deployments that have different Docker and k8's requirements, but are related. A simple example is one deployment loading data to s3 and another deployment running dbt on that data. It would be incredible if I could create a parent flow which triggered the pre-defined deployments sequentially. Some (hacky-ish) solutions I can think of: • Define the logic in one big flow and consolidate to one Docker container • Create a flow with the prefect-shell collection and run deployment CLI commands to trigger the deployments sequential These solutions would definitely work as expected. I am curious if there is another solution using the new python

Deployment

class (not seeing anything in the API reference). Or, if anyone else would find a

.run()

method on the

Deployment

class useful? 😼

HI all, I'm new to prefect, loving it so far. I'm working on a 1.0 project that I've inherited - I was wondering if there is a way to utilize the graphql api to verify that all downstream tasks (including children of children etc) of a particular task run were successful. I can do it via repeated querying for downstream_task info, but it would be nice if there was some other method that didn't involve hitting the api repeatedly. Thanks.

https://medium.com/the-prefect-blog/serverless-real-time-data-pipelines-on-aws-with-prefect-ecs-and-github-actions-1737c80da3f5 I'm trying to use the while true to run a flow. It works locally and keeps on creating new flows like I want. When I run this through the cloud and a helm prefect-agent on GKE, it runs one flow then stops.

Hey all, we’re hosting @Oscar Björhn, a Club42 member, on PrefectLive today at 3p Eastern where he’s going to show us how he integrates Prefect into his workflow, including using different flows with different environments inside a Docker container. Come join us live on Twitch!

Hi there. I have a flow that is running on ECS (Prefect 1.0). I see that some tasks got stuck in running state and they have been there for a long time. How can we avoid this? Can prefect cloud do something about them? Is it possible to set a timeout at the task level? I manually stopped them but it would be good to have an automated way of killing these if they run for more than X hours. Thanks!

Hello - have a quick question regarding the local dask executor: We have stumbled upon this issue as well and we are working with prefect version 0.13.10. So it seems that the examples as suggested in prefect docs for LocalDaskExecutor dont work and we need to go via this RemoteEnvironment variable. Has this been addressed at later versions of prefect or should we just proceed with the RemoteEnvironment solution ?

It looks like create_flow_run is internally looking for a Cloud URL. Is there a corresponding call for Prefect Core that doesn't need a cloud URL?

Hi, I'm curious if there are project count limits on Prefect Cloud?

I keep getting this error when trying to have a scheduled run from a deployed flow. It runs fine as a non-deployed flow (when I run it from my terminal). I built and applied the deployment on another flow/script and it worked (but it also doesn’t have any non-prefect imports). Should I

import psycopg2

from somewhere else instead? Running 2.3.2

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 883, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/tmp/tmpcre52tf5prefect/SCRIPTNAME.py", line 3, in <module>
    import psycopg2
ModuleNotFoundError: No module named 'psycopg2'

Hi, I'm just starting to use prefect 2.0. I noticed the prefect-email collection. Is it possible to call a new flow from a task?

Hi guys, I need super urgent help!! My prefect UI stopped loading suddenly!

Hello! A flow run was created by a nonexistent user. Is this expected?

I have a flow function decorated with @flow and specify a DaskTaskRunner. I thought this would cause my tasks to get run on my Dask cluster but I saw this in your documentation: “To run your task with a task runner, you must call the task with .submit().” Is this still true even if my flow function is configured as described above?

does an agent that services a work queue with concurrency 3 end up running three python threads (single python interpreter)? if so, does each thread have it's own event loop?

Can flow run context be fetched within a task in 2.0?

hi all, is there a way to use a parameter as the name of the folder for a

S3Result

? for a flow level result I want to have something like this

bucket/jobs/{parameter}

Anyone had an issue like this when a Flow is restarted by Lazarus? (it was a Spot instance that got yanked away)

ValueError: dictionary update sequence element #0 has length 1; 2 is required

Hi guys, does anybody knows why the Flow Run Logs in prefect UI doesn't reflect anything after "Beginning execution...." I can see additional log lines in my terminal - but it doesn't show in prefect 🤔

Hey everyone, I keep getting very weird errors when trying to scale my workloads in a Kubernetes cluster. Large amounts of flow runs fail with the following error:

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1103, in begin_task_run
    return await orchestrate_task_run(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1186, in orchestrate_task_run
    state = await propose_state(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1484, in propose_state
    raise prefect.exceptions.Abort(response.details.reason)
prefect.exceptions.Abort: This run has already terminated.

This really tells me absolutely nothing, anyone have an idea maybe? A second error that I see again and again is this one:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py", line 186, in connect_tcp
    addr_obj = ip_address(remote_host)
  File "/usr/local/lib/python3.10/ipaddress.py", line 54, in ip_address
    raise ValueError(f'{address!r} does not appear to be an IPv4 or IPv6 address')
ValueError: 'api.prefect.cloud' does not appear to be an IPv4 or IPv6 address

Which again seems super cryptic to me 😅 Any hints to where I could start digging would be greatly appreciated, thanks 🙂

Hi all 🙂 I am experimenting with a setup running the prefect-server in kubernetes, and having another container run an agent. But before running this agent, using

prefect agent start -q 'foo'

for example, I’d like to create and apply deployments using the

Deployment.build_from_flow

syntax described here https://docs.prefect.io/concepts/deployments/#create-a-deployment-from-a-python-object . Then I want this python script to run before I start the agent…

from prefect.deployments import Deployment
from flows.deployments.example_script import flow_name


deployment = Deployment.build_from_flow(
    flow=flow_name,
    name="Test",
    work_queue_name="test",
)

deployment.apply()

I’ve tried running it standalone and in Docker but I get a Attribute Error

module 'flows.deployments.example_script.flow_name' has no attribute 'name'

If I use

setattr(flow_name, 'name', 'something)'

before the definition of

deployment

then I don’t get this exact error but this

ValueError: Could not determine flow's file location.

I’m unsure of where to begin and would love any sort of help

Hello. My question is not directly related to Prefect 2, but it worked fine with Prefect V1. I am getting the following RuntimeError on import because one of my modules tries to call

asyncio.get_event_loop()

.

RuntimeError: There is no current event loop in thread 'AnyIO worker thread'.

Do you have any idea how to avoid it?

Hello all. I recently updated from Prefect 2.1.0 to 2.3.2. After updating, I re-deployed my existing deployments. I am now seeing flows being scheduluded twice. I do not see duplicate deployments listed in the UI or API. Based on the created dates, it appears Prefect is still scheduling using the 2.1.0 deployments as well as the 2.3.2 deployments. The 'old' flow runs have null value the deployment_id field. FYI I am using Prefect cloud. Has anyone seen anything like this and/or have a recommended way of purging what seems to be previous Prefect deployments still being scheduled?

Hi all, I have a question regarding using great expectations with prefect in production environment, I have an agent running on EC2 and flow code stored in S3. However, whenever I deploy and run a flow that has great_expectation's

run_checkpoint_validation

I get the following error but it runs fine locally. Also, the s3 storage used to store flow code has the great_expectations directory in it..

Hi, I’m having trouble getting prefect annotations to work with asyncio. I get the code to run fine but mypy complains that the annotated functions are no longer asynchronous. Removing either prefect annotations or typings allows it work Example:

import asyncio
from prefect import flow, task


@task
async def slow_task(a_value: int) -> int:
    await asyncio.sleep(1)
    return a_value


@flow
async def my_flow(a_value: int) -> int:
    return await slow_task(a_value)

if __name__ == "__main__":
    result: int = asyncio.run(my_flow(42))
    print(result)

results from mypy:

test.py:13: error: Incompatible types in "await" (actual type "None", expected type "Awaitable[Any]")
test.py:17: error: Argument 1 to "run" has incompatible type "None"; expected "Coroutine[Any, Any, <nothing>]"

Any tips on making typings work with asyncio and prefect

Hi. Why Prefect 2 runs flow in a separate thread? I have an issue with

tornado

that tries to get current event loop and fails.

Hi all, I was looking through the documentation of Prefect but could not easily find the answer to my two questions, so I hope you all can help me. Does Prefect support human interaction in the pipelines. e.g. data is being processed in the pipeline but must be validated manually before the final few steps. And does Prefect support states in the pipelines. Meaning the current run can be changed according to the last 10 runs? e.g. data is being processed and if the current value is greater than X and in the last 10 values there were at least 5 others with a value greater than X then I want the workflow to branch differently. I hope I explained it well enough