Has anyone used a kafka processing library with Prefect? I'm thinking about creating a prefect flow, using Faust to process kafka topics

@Zanie not sure who else to tag, but I will also create an issue on Prefect’s github. Possible code braking bug found with a complete reproducible code included in this thread. Basically, if a dataclass has a

__post_init__

method and checks and initializes a

self

parameter, the

__post_init__

gets called AGAIN when the a task run gets created.

I'm using prefect 2.10 to manage flows on a Kubernetes cluster, but I'm not connecting to any kind of Prefect cloud server. However, I'm starting to see this error:

Failed("Flow run encountered an exception. PrefectHTTPStatusError: Client error '429 Too Many Requests' for url '<https://api.prefect.cloud/api/accounts/XYZ/workspaces/XYZ/flow_runs/XYZ>'\nResponse: {'detail': 'Orchestration API rate limit reached'}\nFor more information check: <https://httpstatuses.com/429>\n"

I don't know how this is possible - since I'm in no way interacting with prefect server. I've used previous versions of Prefect and never seen this before. I've replaced the specific strings with XYZ

I’m getting an error that’s logged when running async flows locally with prefect that complains:

Exception ignored in: <async_generator object add_event_loop_shutdown_callback.<locals>.on_shutdown at 0x7f3d81b374c0>
Traceback (most recent call last):
  File "/home/jon/miniconda3/envs/abraxas-env/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 308, in on_shutdown
AttributeError: 'NoneType' object has no attribute 'pop'

Which seems related to this issue (which I’ve already commented on). I have no idea how to solve this. It seems like some odd threading/multiprocessing bug where EVENT_LOOP_GC_REFS is not being copied over correctly somehow.

Considering code like this.

def child_flow_c():
    d = run_deployment("child-flow-d/dep-child-d")
    # How can I be sure that run_deployment("child-flow-d/dep-child-d") is completed before moving to next line/deployment 
    e = run_deployment("child-flow-d/dep-child-d")

How we check the status of

run_deployment

is completed before executing another

run_deployment

Hi Folks, my team and I are moving over some pipelines that make use of S3, and I cannot figure out how to get this new AWS creds block or the new S3 block working. Has anyone gotten this working? I've tried bringing over all sorts of configuration options, but I keep running into "profile not found for profile _____" and access denied errors. I can connect just fine with boto3 directly, but would like to use these blocks to shorten my code, but at this point, I feel like these blocks are more likely to shorten my lifespan 😆

I'd like to bulk cancel or delete scheduled flow runs but don't know how to in prefect cloud

I upgraded to the latest

prefect==2.10.11

version because of sqlalchemy 2.0 support. Thanks for the fast support there! 🚀 Now I have a problem with my custom block when I try to run my flows:

KeyError: "No class found for dispatch key 'my-custom-block' in registry for type 'Block'."

prefect_aws==0.3.2

is installed and

PREFECT_EXTRA_ENTRYPOINTS

is set. I’m using prefect cloud and the agent is updated to the latest version. I downgraded my prefect docker image to version 2.10.10 and it works fine again. Full log attached in the thread

Hi everyone, I have a question regarding the priority queue in the workpool So let suppose i have 2 queue (q1, q2) in a workpool (w) where q1 in a priority queue with highest priority. Now let say i have to multiple input (input1, input2 and so on) where "input 23" has the highest priority and need to processed as soon as it came to "q1" My question here is: 1. The scenario i mentioned above is okay ? 2. if input 16 come to q1 then all the inputs which are in q2 will be put on to hold till input 16 of q1 is processed ? Can anyone please help me on this.

been getting

Orchestration API rate limit reached

all day, any idea what's up? I have mapped functions that around 3-4k iterations. by the time I am around half of it. I hit this error and it crashes on its own. When I retry, by the time it finds all completed tasks (checkpoints), it will raise the error again and won't move forward. using prefect:2.10.10-python3.10 and running on ecs with s3 result storage

Is there any reason for the release of Prefect 2.10.11 not being announced in #announcements?

is there any prefect-dbt channel?

Hi everyone, Whats the difference between a variable and a StringBlock?

umm if any1 is an idiot like me n hadn't specified sqlalchemy version for their images. Now would be a good time, upgrading to prefect 2.10.11, sqlalchemy v2 was installed instead of v1

When using projects for multi-deployments, it looks like I should be able to have my deployment.yaml file in the parent directory of all the project folders; however, when I try to deploy I get the following error (i’ll put it in the thread)

One of my flow module imports itself imports our SQLAlchemy ORM models. When deployment.build_from_flow is run it throws this error (posted as a reply here). SQLAlchemy version is 1.4.46 Prefect version is 2.10.11 Python version is 3.11.2 I have tried decorating the

class AnnotationAirtable(Base)

declaration with typing.@runtime_checkable, but received the same error. I have tried separating the flow import with another module as a buffer (in desperation, to try to stop the ORM model import from happening!), also to no avail.

Hi everyone, I have a Prefect task that starts a child process that can take a while to run. When I cancel the flow, the child process doesn't get killed with it and this takes resources for nothing. Is there a way to kill it?

Hello. In Prefect2, I have a use-case where I need to share an object between 2 flows: Flow A instantiates a class and if the flow fails, I want to persist that object so that Flow B (started manually by a person) can retrieve that object and perform an operation. I'm thinking of doing this using result persistence, but I couldn't find a way to retrieve a flow run's persisted result from another flow. Any ideas on how to do this? Thank you

I’m still running into a lot of issues with async flows — raised this issue 10 days ago specifically related to subflows - but having massively stripped down the amount I’m using prefect (basically just using it to run the

main

function - with everything else being plain async python) and i’m still getting random hangs — and when I kill the script, it seems (as best as I can tell), that the

KeyboardInterrupt

gets caught somewhere within the prefect codebase. what is curious is that when I dropped the top-level

@flow

altogether and just ran as a plain python script, it hasn’t stalled yet (will update if it does). is it possible that there’s something going on with longer-running (5min+) async flows that’s causing them to hang at some point? I thought maybe it was to do with hitting the logging rate limit but by removing all of the task decorators I’ve managed to limit the overall prefect calls significantly. I’m racking my brain at this point. anyone have any experience with long-running async programs? are there other factors that could be causing random hangs?

When querying the

api/work_queues/[ID]/status

endpoint, is it possible to modify the health_check_policy? Here is a typical response:

{
  "healthy": true,
  "late_runs_count": 0,
  "last_polled": "2023-05-09T17:08:54.625892+00:00",
  "health_check_policy": {
    "maximum_late_runs": 0,
    "maximum_seconds_since_last_polled": 60
  }
}

Hi Team! Is there a way we can get prefect logs to also pick up logs from a .txt file? What's essentially happening is that we are using Prefect to kick off another job, which logs to a .txt file, but we want prefect to pick up those logs in real time as the process executes. I understand there's a logging.yml, is there a parameter there that could help us?

Is there a good reason calling sub-flows and/or

run_deployment()

doesn't expose a

PrefectFuture()

interface similar to

Task.submit()

? For a flow like:

@flow
def f():
  a = subflow_foo.submit(1)  # .submit() doesn't exist on Flow or run_deployment().
  b = subflow_foo.submit(2)
  c = subflow_foo.submit(3)

  x = subflow_bar(a, b)
  y = subflow_bar(b, c)

  x.wait()
  y.wait()

I want to get the engine's dependency tracking to kick off downstream sub-flows as soon as their dependencies are satisfied. It feels like I could wrap run_deployment() in a future myself, but maybe I'm missing some non-obvious reason this wouldn't work?

I can't find much other than a thread from 2021 (https://github.com/PrefectHQ/prefect/issues/4789) that describes this problem, but is there some way within the snowflake block's methods to get the column names coming back with snowflake query results?

Can anyone help me out with how to manage dynamic dependencies in prefect flows? I'm trying to create a pipeline in which a process is triggered and a python script is executued. Everytime a python script is submitted, first its dependencies should be installed, and then the script should be executed. How can I handle dynamic pip packages in this context? Tried to run prefect by docker in docker but the overhead is too much both from networking PoV and also resource wise. Currently trying to go for python venvs + a way to manage pip dependencies.

Hello all, When deploying flows from a python using an RRule scheduler to schedule daily updates, I’m consistently seeing odd behavior where a flow gets duplicated many, many times, 1-3 seconds apart. See below screenshot — ~100 more flows for 3 days follow, when I expected to see only 3. Any ideas what might cause this? The RRule scheduler in question is

FREQ=DAILY;BYHOUR=10;BYMINUTE=21

. Even more curiously, sometimes the schedule will seem to work correctly immediately after running. Then I will check back some hours or a day later and the duplication will have happened.

any1 else had issues where they are running agent on aws(fargate) and the flow is in running state but there isn't a task running in was anymore? I don't realise it unless I check the flow logs and notice that the last log was probably an hour ago or so.

Hello, I've set up an Automation in Prefect Cloud to send a slack notification when a work queue becomes unhealthy. I've used the same block (Slack Incoming Webhook, from the prefect_slack pip package) I have stored to send other slack notifications within the flows, which works fine. However, when the automation is triggered it fails with the following error:

The notification block was invalid: KeyError(\"No class found for dispatch key 'slack-incoming-webhook' in registry for type 'Block'.\")

Is this a bug?

Hello, I have one doubt, If we use a

task

with

cache_key_fn

, then that where is that cache is stored in Prefect cloud? If I use self hosted version then where this cache is stored and retrieved?

Hi there, I’m playing with Prefect Python SDK. It seems the

task

decorator is causing some issues with

mypy

. 🧵

Hello there I try to use new concept of projects and to deploy with

prefect_docker

template to build a docker image and to push it on my private registry (on Gcloud) but it looks like it's not possible to use an other registry than docker. Any idea? Otherwise, what's the best practice to deploy a docker image flow on a private storage?