Bonjorno friends, I have been trying to make the prefect cloud work for the past few days. After resolving various errors and deployment crashes, I encountered a new issue: the agent is raising a NotImplementedError. To address this, I followed the suggestion in an article I found and updated to Python 3.11.5. However, I am now encountering the following error:

(anton) C:\Users\Anton\Desktop\DE2\2_workflow_orchestration\flows\03_deployments>python flow_pandas.py  
Traceback (most recent call last):
  File "C:\Users\Anton\Desktop\DE2\2_workflow_orchestration\flows\03_deployments\flow_pandas.py", line 5, in <module>
    from prefect import flow, task
  File "C:\Users\Anton\anaconda3\envs\anton\Lib\site-packages\prefect\_init_.py", line 37, in <module>
    from prefect.runner import Runner, serve
  File "C:\Users\Anton\anaconda3\envs\anton\Lib\site-packages\prefect\runner.py", line 46, in <module>
    from prefect.client.orchestration import get_client
  File "C:\Users\Anton\anaconda3\envs\anton\Lib\site-packages\prefect\client\orchestration.py", line 11, in <module>
    import pydantic
  File "pydantic\_init_.py", line 2, in init pydantic._init_
  File "pydantic\dataclasses.py", line 41, in init pydantic.dataclasses
    # +=========+=========================================+

ModuleNotFoundError: No module named 'typing_extensions'

I updated all the packages in my env, prefect, python, etc. checked all the possible solutions around 'typing_extensions'. Maybe I missed something obvious? Do you have any suggestions to where should I look? Cheers!

Would Prefect (cloud) be a good choice for a "queue of async logic" type of workload? It's not a typical scheduled data ETL pipeline scenario but instead it's small parts of logic that are triggered by actions of users of your app. For example, in our app, when a user purchases a coupon we want to start a flow that performs some logic (some DB and API calls) in a durable fashion. Specifically, we would use the ECS push work pools. So let's say tens, hundreds of new ECS tasks per minute. I'm wondering if this is a viable use case of Prefect (performance and cost wise), compared to something more traditional like, let's say, a kafka topic with a consumer that performs this logic asynchronously? Thanks!

Hello again! 'm relatively new to working with Prefect, and my current project involves integrating it into my Kedro pipeline. Unfortunately, I've run into some challenges when attempting to run it within a Docker environment. I've been struggling to locate any documentation that specifically addresses the integration of Prefect into a Kedro pipeline project. It seems that implementing Prefect within a Kedro pipeline may require a distinct approach compared to using it in regular Python scripts. I would greatly appreciate any guidance or insights you can provide. Thank you in advance! Also, I'd like to share an additional experience as well. In the past, I faced connectivity problems with Prefect, even though I had started the Prefect agent. I was unable to establish a connection to both the local server and cloud server instances. To resolve this issue, I resorted to starting the entire process from scratch. I'm wondering if there might have been a specific step I missed or if there's something I may have done incorrectly that led to this problem. Your insights on this matter would be greatly appreciated. Thank you once again!

I am working on a Custom Worker and am getting this error in my flow logs

Infrastructure returned without reporting flow run 'ae92eb90-d347-4c4d-b15b-ad4cbf3b5466' as started or raising an error. This behavior is not expected and generally indicates improper implementation of infrastructure. The flow run will not be marked as failed, but an issue may have occurred.

I am returning a 0 status_code and an identifier, so I can't figure out what is triggering this

Good day. I am running into some behavior with the prefect CLI that I do not understand. I have a profiles.toml with local, eks, and cloud entries. Even if I specify

prefect profile use 'eks'

or

prefect profile use 'local'

,

prefect deployment ls

returns the list of deployments from my cloud instance. This also occurs with

prefect -p 'eks' deployment ls

or

prefect -p 'local' deployment ls

.

prefect profile ls

shows all three profiles and shows 'eks' as the active profile. The deploy commands also seems stuck on the cloud profile. Any ideas what's going on?

I was finally able to set up a prefect worker service in my ECS, and I defined an execution role ARN in the task definition. However, I'm still getting the following error when I submit my flow run:

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.

So, in Prefect 1 we would attach schedules to flows with specific arguments (ie, "run with X arguments every day at midnight, run with Y arguments every week"). How do we do this with a Prefect 2 Deployment? Does each one have to have a unique name?

Hello all. Im trying use prefect with my new project, but has some problem with dependency . depends on both pydantic (^2.4.2) and prefect (^2.13.4), version solving failed. prefect use pydantic < 2.0.0. Anyone knows how can i make some workaround about it ?

Hi! I would like to attach some result metadata as key-value pairs to each flow run, so that I could capture some basic metadata like “# of rows copied” or “# of rows in table” and stuff like that. Preferably I would not use results mechanism as having an external storage location feels like an overkill. Is there a way to achieve this?

hi, when using prefect server i'm getting a restart loop with this error:

File "/usr/local/lib/python3.10/site-packages/alembic/script/base.py", line 276, in _catch_revision_errors
    raise util.CommandError(resolution) from re
alembic.util.exc.CommandError: Can't locate revision identified by '05ea6f882b1d'

Application startup failed. Exiting.
Server stopped!

full stack trace in thread happened to anybody else?

Hi, I am using Prefect Cloud, I created one deployment for one work-pool which has one work queue. Now in kubernetes, I created a Process Worker pod and I scaled it to 10 pods horizontally. I put 10 as limit of flows should run on each worker. Now if I create 100 flows, so technically, all 10 prefect worker (process type) pods should pick 10 flows and execute in paralelly. But that is not happening. There is any guide how to scale workers on single work-pool to execute flows parallelly?

Hi, I’m using prefect 2.13.0 on windows and noticed that if I try to retry a flow from the UI (either after a task has failed, I’ve paused the flow, or I’ve cancelled the flow) it will retry all tasks, even those that were in the “Completed” state. Any ideas how I can get the flow retry feature to only retry non-completed tasks?

Hi all! We've recently noticed our prefect postgres database has hit 20GB in size, which seems like a large size. Does anyone know if prefect is saving out task results / artifacts (note I have no explicit artifacts in any of my flows), and if there are ways to configure a retention policy (like to delete things after a week)?

Hey team, is there a prefect 2 equivalent for workers of the prefect 1 agent start

--no-pull

?

Hey everyone! I have a dataframe that is being passed between tasks in the flow and by using cache it is much faster and there is no need to persist it to disk. My question is, is it possible to clear that big dataframe from cache completely after the flow is done and not by setting cache_expiration? I read into refreshing the cache but that did not help since I do not repeat tasks. I also tried with

del dataframe
gc.collect()

But it does not get released from cache Any ideas?

Does anybody have a link to Server

Notifications

documentations. The link from the UI is sending me to 404 error. I would like to customize my slack message, however I can not understand how to extract flow data inside the configuration setup

Good morning and happy Monday. I have a quick questions in regards to Prefect 1.4. Is there a way to cancel alot of flows that are running without the usage of UI (or thru the UI). The number of scheduled flows are substantial so it's not possible to cancel them one by one thru the UI.

I am trying to run a Prefect flow in ECS with a prebuilt docker image. However, when the container is pulled the default command is being run; where would i configure the prefect flow entrypoint to execute? Also, what is happening under the hood when the flow run is executing? Particularly with regards to the parameters sent, what is organizing that and how is that information sent to the environment where the flow run is being executed?

Hi, everyone. I followed this tutorial () and was able to set up Prefect agent in ECS. However, I would like to use prefect server, instead of prefect cloud. Is there a tutorial or learning resource where I can learn how to setup prefect server in ECS? p.s: thanks @Taylor Curran and Mat for the tutorial. You did a great job!

hey folks... we're trying to suppress distributed.worker and distributed.core from our datadog logs, and have traced them to prefect dask. can anyone share how we might suppress any dask-related logs in prefect? we tried specifying the loggers to suppress by name and it hasn't worked...

Is there a channel for feature requests? Now that I have everything set up in ECS I have some suggestions, particularly around the build control plane 🙂

When I set up Prefect in ECS, I went ahead and created the prefect worker task definition, but I noticed that there is also a prefect task definition for the Prefect flow run execution. Was that created by Prefect?

I am seeing a weird issue with tasks and pydantic models. This is my example. The exclude_unset=True is not working once when the task is called. If I use user_json.fn() it works as excepted. Does anyone know why the @task decorator would be messing with the pydantic model and if a known workaround is available?

class User(BaseModel):
    """Pydantic model for the User."""

    email: Optional[str]
    firstName: Optional[str]
    lastName: Optional[str]


@task()
def user_json(user: User):
    """Should return a JSON representation of only the set fields in the User model."""
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"email": null, "firstName": "John", "lastName": null}
    return user.json(exclude_unset=True)


@flow()
def fl_create_user():
    """Creates a new user with only the first_name set and calls the user_json task."""
    user = User(firstName="John")
    print(user)
    # Outputs: email=None firstName='John' lastName=None
    print(user.json(exclude_unset=True))
    # Outputs: {"firstName": "John"}
    json_user = user_json(user)
    print(json_user)
    # Outputs: {"email": null, "firstName": "John", "lastName": null}

I'm running dependent subflows with

run_deployment

(run from a task). I notice that if one fails, then the downstream ones don't fail. How do I make subflow failures propagate to the parent flow? I'm using the

ConcurrentTaskRunner

. Some need to run concurrently with each other.

for an example: I want the ones without the

wait_for

argument to all run at the same time, while any Failure kills the whole thing.

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
import time


def fail_hook(flow, flow_run, state):
    print("Failed")
    raise Exception(state.message)


@task(log_prints=True, on_failure=[fail_hook])
def my_task():
    time.sleep(15)
    # some code here
    raise Exception("This task has failed.")  # this will set the task to Failed state


@task(log_prints=True, on_failure=[fail_hook])
def my_task2():
    time.sleep(25)
    return "sdfsdf"


@flow(
    task_runner=ConcurrentTaskRunner(),
)
def my_flow():
    result1 = my_task.submit(return_state=True)
    result2 = my_task2.submit(return_state=True)
    result3 = my_task.submit(return_state=True)
    result4 = my_task2.submit(return_state=True, wait_for=[result1])

hi all, i have a question about whether prefect is the tool I'm looking for. I'll post the description inside the thread - it's on the longer side, looking forward to hearing your thoughts marvin

We are running prefect for 90 days in production. Today we noticed that the Prefect Server is querying Postgresql at a rate of 14MB/s. Grew from 3MB/s linearly over a course of a month. Is this something you are aware of? We nuked the db now and checking if this is related to the number of flows we ran so far.

Hi All, i am a newbie i have setup the dev environment when i am trying to run the command "prefect dev start" i am able to go to the local host link.But i am not table to see any UI it is showing blank page for me. Please help me on this. Thanks,

I am using this in debug mode

Are there examples of using 1 flow to run 2 subflows in Prefect 2.0? The documentation is not very clear on the matter. How do I test it locally? On my firm's local server?