We just had an edge case with Prefect 1 (v0.15.6) self-hosted, where two agents which started with the same tags started exactly at the same time a flow_run. As we are moving to prefect 2, can multiple agents listen to the same work queue ? this would be for self-hosted on which we have multiple instances of Orion.

Hi, I’m trying to sending alarm to slack when my flow failed. Normal case it works well, but when flow about lack of aws resource (i.e., gpu) fails i didnt’t get the alarm. I’m using ECS agnet now. Anyone could help? Thanks.

Hello everyone. This more of a general question than a specific issue. I have a set of requirements and I'm unsure if Prefect is suitable for them (perhaps ETL tools in general aren't). I need to aggregate data from different sources, process it in chunks, and then stream the chunks to a browser for each client. I think conceptually there are 2 ways to map this onto Prefect. (1) Have tasks for aggregation, and then a separate long-lived task that handles chunking, processing and streaming. In this case one flow takes care of the entire job. (2) Have tasks for aggregation, then a subflow for each chunk, where we have tasks for processing the chunk and a separate one for streaming it back. Both approaches do something that maybe Prefect isn't designed for, I don't know enough about Prefect's internals to understand the performance/space cost of a single flow, or the potential problems of a really long-lived task. (1) would have a really long-lived task, maybe even days. (2) Would have huge amount of subflows (there could be around 50,000 chunks) for a single job. I prefer (2) just because (1) feels like it doesn't have the task granularity to really take advantage of Prefect. I'm open to alternative approaches. Ordinarily I'd reach for something like Kafka for this, but I like getting the benefits of Prefect (logging, observability, retries..etc) for free.

hi everyone! im facing a weird issue in flow runs . im running prefect on aws eks ( self hosted instance of orion and kubernetes jobs on different namespaces ) 1. i have 9 running flows which i can see on the ui but i cant find the corresponding job on my cluster 2. when i check the concurrency-limit for the tags that these flow use i can see 14 task runs ( these flow have parallel task runs so some of them are having more than one task stuck in running state) so basically 9 running flows with 14 running tasks as i can see on the ui but i cant find the jobs on my cluster. doubt - how did these flow runs get terminated on the cluster and even if they did why did it not update in the ui or lets say why was the state not updated in meta db attaching screenshots for the same

Hello, Can I get the input arguments of a task-future?

For our setup, we're incorporating unittest in the build phase of our ci/cd pipeline. So that it won't deploy it to the cloud, unless the test passes in Azure devops I'd like to be able to run unittest on singular tasks and not start a flow run but when running

import unittest
from tasks import address_matching as am
from prefect import flow,task


class TestDatahandling(unittest.TestCase):
    
    def test_tilslutning_forsyning(self):
        am.address_matching()



if __name__ == '__main__':
    unittest.main()

I get following error.message

ERROR: test_tilslutning_forsyning (__main__.TestDatahandling)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "c:\Users\nho\Desktop\git\Prefect\etl\geoserver\tilslutning_forsyning\test.py", line 9, in test_tilslutning_forsyning
    am.address_matching()
  File "C:\Users\nho\AppData\Local\Programs\Python\Python310\lib\site-packages\prefect\tasks.py", line 353, in __call__
    return enter_task_run_engine(
  File "C:\Users\nho\AppData\Local\Programs\Python\Python310\lib\site-packages\prefect\engine.py", line 674, in enter_task_run_engine
    raise RuntimeError(
RuntimeError: Tasks cannot be run outside of a flow. To call the underlying task function outside of a flow use `task.fn()`.

Two questions 1. What would the syntax for task.fn() in my example be? I can't get it to run 2. Are there better ways, or best practices of running tests on flows or does this setup seems okay?

Hello everyone , i am using prefect and dask resource manager and running some tasks within Dask resource manager context and few outside the context manager. Flow looks like this

with Flow("Test flow") as flow:
   
    with DaskCluster(n_workers=n_workers) as client:
        data = extract()
        processed_data= transform(data)

    save_data(processed_data)

The problem is dask cluster is not shutting until save_data function completes but i expected that as soon as

transform

function completes , the

cluster cleanup

should happen, Is there any way i can initiate

save_data()

after

transform()

is done and the

cluster_cleanup

is done as well.

Hi Prefect Community, I am running a machine learning model as a workflow using Prefect. If I don't use Flows and Tasks, the same model completes within 90 minutes. But if I convert them as Flow and Tasks, the same model takes 300 minutes to complete. I think, especially it takes more time in dropna() or drop() functions (probably more than that). Is there any way I can allocate more memory for each task to make it run faster since it is a memory-based process? Or do we aware of any memory allocation restriction while running a task or flow? Please shed some light. Even I tried with DASK parallel processing, it didn't help when we run it with a Flow and Tasks. BTW, I am using Prefect Orion 2.4.5 and in a on-premise Linux server.

Hello, I'm trying to implement the following kind of logic within a Prefect1 flow.

@task(name="Task A")
def task_a(x: int) -> int:
    if x == 2:
        raise signals.FAIL
    return x + 2

@task(name="Task B")
def task_b(x: int) -> int:
    return x + 2

@task(name="check")
def checker(x: int) -> None:
    if anything_upstream_broke():
        this_didn't_work(ID)
    else:
        this_was_ok(x)


with Flow("x example") as flow:
    l = [2,5,7]
    la = task_a.map(x=l)
    lb = task_b.map(x=la)
    lc = checker.map(x=lb)

That is, I have a list of things I want to run through the WF, and I

map

these. Sometimes some of the elements in the list won't run properly. I'd like a way to look through all of the upstream tasks and check if any failed and do a thing for that specific input.

Hi! What is the best way to create a scheduled flow with multiple scheduled subflows? Couldn’t find an article or discourse discussion. We are using Prefect Cloud 1.0. Example of structure: • Main flow A, run every week, dependant on: ◦ Sublow B, run every day, dependant on: ▪︎ Sublow C, run every hour So, if any of subflows fail, all upstream subflows should not start. Is there a way to do so? cc @Taylor Curran

Is it safe to delete the ~/.prefect/storage folder? mine has ballooned to over 100 GB now. anyway we can set some settings to limit how large this folder can get?

What is the Prefect 2 equivalent of

apply_map

? Details in 🧵

hi guys, I’m trying to build and apply a deployment from our gitlab-ci pipeline, but I got

httpx.ConnectTimeout

error, my prefect-orion runs on our own k8s and should be reachable from our gitlab-ci pipeline, has anyone seen this error? any idea how I could debug? Thanks!

Is it safe to stop the prefect agent in order to update the packages installed in the virtualenv? Or would that cause any flow-runs-in-progress to be killed/restarted? Asking for both prefect 1.x and 2.x, as we are using both.

Hi, I'm creating a flow-of-flows and was wondering if anyone had any recommendations for my scenario. Background: My parent flow has potential to create a lot of child flows, with the precise number (N) determined at run-time. To handle this, I've been using mapping. However I've noticed

create_flow_run.map

will create all of flow runs at once, which is clogging up my agent's backlog when N gets too large. This creates a lot of failures, which occur before my child flow runs can even begin. Unfortunately, I am not the admin and don't have control over the agent we're using, so I don't have a way to easily investigate why this is or resolve it. Even if I was, it seems like kicking off this many flows at once probably isn't a good idea. Thus, I'm wondering if there is some way to throttle

create_flow_run

my flow without pre-defining my N child flows. I'm even OK with running all N flows in sequence. I.e. kick off flow run 1 -> wait for flow run 1 -> kick off flow run 2 -> etc. Here is the structure I'm using now:

def create_wait_on_child(parameters):
    create_id = create_flow_run(flow_name="GenericChild",
                                project_name="Project",
                                parameters=parameters)
    return wait_for_flow_run(create_id, 
                             raise_final_state=True)

with Flow('Parent Flow') as flow:
    flow_runs_params_list = get_flow_runs_params()
    apply_map(schedule_run_backfill, flow_runs_params_list)

My Question: Is there some way to combine the functionality of`create_flow_run` and

wait_for_flow_run

in the same task? I know I can't just add a

task()

decorator to

create_wait_on_child

in the example above (since that would involve tasks with tasks), but that is the functionality I'm going for. Or does anyone have alternative recommendations? Appreciate anyone's input.

Two questions. What is the collective noun for people who use Prefect? Is there a command that tries to communicate with an Orion server in a "hello world" sense to purely test for connectivity? I would like to perform an initial test up from before starting my python workflow. I have a self-hosted Orion server and the aim is purely to make sure this Orion server is accessible.

Is there some general logic for what code should be @prefect.task'd and what code can be left in the @prefect.flow without tasks? (talking strictly prefect 2)

Hello everyone , i am using prefect and dask resource manager and running some tasks within Dask resource manager context and few outside the context manager. Flow looks like this

with Flow("Test flow") as flow:
   
    with DaskCluster(n_workers=n_workers) as client:
        data = extract()
        processed_data= transform(data)

    save_data(processed_data)

The problem is dask cluster is not shutting until save_data function completes but i expected that as soon as

transform

function completes , the

cluster cleanup

should happen, Is there any way i can initiate

save_data()

after

transform()

is done and the

cluster_cleanup

is done as well.

Hi all, in Prefect 2 we are using environment overrides in our deployments. However I cannot see these in the UI. Does anyone know if these can be made visible? If not, are there plans to show these overrides in the UI? thanks!

Hi. What is correct approach to mock Blocks in tests(sync and async version)?

“see Dask worker logs from the Prefect UI.” YES!

Hi, is it possible to add dependency for a subflow, so it will wait for a task to complete? Similar like wait_for when using a task.

This message was deleted.

Hello, I am running into an

asyncio

exception when using

DaskTaskRunner

which leads to tasks hanging and not completing. -->

In Prefect Live at 3PM Eastern today, Allan Campopiano, a data scientist at Deepnote, will talk about ML learning in the warehouse with Snowpark and Deepnote. Come join us live on Twitch or YouTube.

Hi all, regarding

PREFECT_LOGGING_EXTRA_LOGGERS

Has it to be set in the flow, on the agent or on the orion server ? Thanks

Hey everyone! I'm having some problems with my agents (2 agents) , they are not submmiting any flows which has caused a late in my deployments (currently in 2513 Late runs) They are deployed in a EC2 with systemctl. I restarted them, installed last version of prefect (2.6.9) and the logs only shows that (no submission or any error).

Agent started! Looking for work from queue(s): xxxxx...

Don't know what else I can do 😅 Thanks!

I am encountering

Permission denied

errors with the github storage blocks since today. Has anyone seen this as well? The error appears if the github repo is already present from a previous run. If the repo is deleted the gitrepo is cloned without any error.

Hello Prefect community. We are getting an

Unknown Opcode

error in our prefect flow. I looked up the error and saw several issues relating to mismatched python versions. I have verified that we are using python 3.8 (3.87 and 3.8) for everything. Does anyone have any ideas why this error may be occurring? Other tasks in our flow ran successfully. Only one failed.

Hey @George Coyne thanks for your help. One last question for now, the deployment yaml file is now in the directory of all my code that is used to build the image. Will moving it cause any issues? Does it need to be in a certain place?