Hello. In prefect2, I have a few upstream/downstream subflows that I want to run for n inputs in parallel, but for each n, the dependency should be respected (e.g., if subflow_1 fails for n=x, subflow_2 should be skipped for n=x). What is the cleanest way to implement this and using which task runner? (I'm thinking of defining a main_flow that takes n as input and manages the dependency of subflow runs, so I just run the main_flow in parallel for n) Thanks

Feature idea: Looks like there isn’t a way to spread the work load between many prefect agents: https://discourse.prefect.io/t/agents-and-work-queues-faq/1468#what-are-use-cases-where-starting-multipl[…]r-the-same-work-queue-can-be-beneficial-4 Currently I have many agents working on the same queue and it happens that 1 agent fetches many flow runs and unbalances the work load, threatening sometimes to use more resources than it’s host machine has. My current solution for this is using the

--limit

flag in the agent, but that’s not ideal. I though it would be nice to have a

--prefetch-limit

option to the agent (beyond the already present

--prefetch-seconds

). This way, an agent could have a big

--limit

but not fetch too many flow runs at once, therefore spreading a bit the load to the other agents. What do you think?

I am encountering a new issue since this morning. Wondering if anyone had the same issue. Using .submit seems to cause the flow to crash. I took out the task that the flow was crashing at, created a separate flow that just contains that task and the tasks and function that it needs. It seems to work, and not crash even when using .submit. But for some reason in the "big" flow, the task crashes when using .submit

Hi there! I'm running into an issue with api authentication. I've sucessfully logged in through my terminal, pasting my API key when prompted but when I try to run a flow locally I get the following error:

Cannot create flow run. Failed to reach API at <https://api.prefect.cloud/api/accounts/a9c0f124-ca06-4646-a501-57a405ebf3c7/workspaces/43f1b8a7-ed9c-46d2-a88a-55eef95b8ef7/>.

-- does anyone have any recommendations? TIA!

In v2, what are the best ways to set dependencies between tasks? In v1 we called

flow.set_dependencies(task_4, upstream_tasks=[task_2, task_3])

In v2 what is the easiest way to send a customized Slack notification if any of the tasks fail?

Hello, I am trying to obtain a future object from a toy task with a timeout state. Then ultimately send that timeout to slack. The slack integration is working but the flow calling the failed tasks never gets to access the timeout state and subsequently send it to slack. the logger prints: uuid-123 the number 42 doesn't get logged probably because it times out but I also never get to log the PrefectFuture object state .e.g. 'Timeout' Is there something I should be doing differently to catch and handle the timeout before the flow Fails

@task(timeout_seconds=5)
def test_task_timeout(uuid: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(uuid)
    time.sleep(10)
    return 42


@flow(timeout_seconds=60)
def test_flow_timeouts(station_id: str):
    logger = get_run_logger()

    future = test_task_timeout.submit(uuid='uuid-123')
    result = future.result()
    <http://logger.info|logger.info>(result)
    flow_context = context.get_run_context()
    prefect_state_handler(future_object=future, context=flow_context)

Hello, I got a question which I can't seem to be able to find any hint about. I can't tell if it's a weird use case that nobody else cares about or if it's so commun that nobody's talking about it ^^ Please excuse me if it sounds dumb. Can I have tasks in a flow executed on different python interpretors ? (not the whole flow but each task in a single flow) We are using different apps to process different things, and many of those apps embed their own python interpretor to provide functionnalities (as opposed to being a python package you can intall). So we would need flows that jumps from python to let's say Autodesk Maya, then pure python again and then The Foundry's Nuke, for example. How can we achieve that with prefect ?

Hi Everyone, sorrect me if i am wrong but prefect 2 does not has a skip state feature, so i was trying to implement the same using conditional flow using if else. for example here is my snipped of flow

@flow(name="Inferencing")
def start_inferencing(my_param, file_path):
    job_id = my_param
    algo_id = "402"
    print(">>>>>>>>>>>>>>>>>>>>>>>>>>>")
    print(algo_id, job_id, file_path)
    print(">>>>>>>>>>>>>>>>>>>>>>>>>>>")
    res_1 = generate_jobID.submit(algo_id, job_id)
    print("\n\n\n\n ***************************\n\n\n\n ", res_1, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_1)
    res_2 = get_file.submit(file_path)
    print("\n\n\n\n ***************************\n\n\n\n ", res_2, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_2)
    res_4 = choose_valid_file.submit(prev_task=res_2)
    print("\n\n\n\n ***************************\n\n\n\n ", res_4, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_4)
    res_5 = prepare_request_upload.submit(prev_task=res_4)
    print("\n\n\n\n ***************************\n\n\n\n ", res_5, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_5)
    res_6 = send_data_request.submit(prev_task=res_5)
    print("\n\n\n\n ***************************\n\n\n\n ", res_6, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_6)
    res_7 = prepare_predict_request.submit(prev_task=res_6)
    print("\n\n\n\n ***************************\n\n\n\n ", res_7, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_7)
    res_8 = send_predict_request.submit(prev_task=res_7)
    print("\n\n\n\n ***************************\n\n\n\n ", res_8, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_8)
    res_9 = extract_lunit_jobid.submit(prev_task=res_8)
    print("\n\n\n\n ***************************\n\n\n\n ", res_9, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_9)
    res_10 = prepare_fetch.submit(prev_task=res_9)
    print("\n\n\n\n ***************************\n\n\n\n ", res_10, "\n\n\n\n ****************************\n\n\n\n")
    call_on_failure_if_failed(res_10)
    res_11 = fetch_request.submit(prev_task=res_10)
    call_on_failure_if_failed(res_11)
    print("\n\n\n\n ***************************\n\n\n\n ", res_11, "\n\n\n\n ****************************\n\n\n\n")
    if res_11:
        res_12 = convert_output.submit(prev_task=res_11)
        call_on_failure_if_failed(res_12)
        res_13 = zip_file.submit(prev_task=res_12)
        call_on_failure_if_failed(res_13)
        res_14 = send_to_HGW.submit(prev_task=res_13)
        call_on_failure_if_failed(res_14)
    else:
        send_log(job_id=job_id, header="AI Gateway", subheader="Failed to Receive Inferencing Result", status="failed",
                     level=5, direction=True, send_ip=SEND_IP, msg_id=8)
        

if __name__ == "main":
    start_inferencing_402(parameters=dict_)

the function res_12, res_13, res_14 i want to execute only if the res_11 gives True, By the way every function returns either true or false, Problem: 1. Not even one print statement is execuiting, 2. Also even if the res_11 is false then also res_12, res_13, res_14 is getting executed, Can anyone please tell me what i am doing wrong.

does anybody have an example working Dockerfile with prefect and dask for GCPClusters?

is there a way, programmatically, either with prefect cli or within python flow, to access the last time that this very flow successfully ran? I'd like to pass in as parameter the last successful run of my flow, so that I only fetch data from last successful run to now.. are there any examples of how to do this anywhere? TIA

Hi guys, I seem to have a problem with getting an ECSTask on EC2 to work. I got to the point when a simple hello world container ran without problems, but when I tried to execute the same flow again, it failed - several times. Meanwhile, the same ECSTask with the same parameters succeeds with Fargate. I am using

g5.4xlarge

EC2s in an an ASG with an amazon ecs-optimized linux gpu AMI. Here is how I am calling ECSTask:

ECSTask(
    command=["echo", "hello world"],
    image="<http://xxxxxxxxxxxx.dkr.ecr.eu-central-1.amazonaws.com/my_repo:my_image|xxxxxxxxxxxx.dkr.ecr.eu-central-1.amazonaws.com/my_repo:my_image>",
    launch_type="EC2",
    cluster="my-cluster",
    stream_output=True,
    execution_role_arn="arn:aws:iam::xxxxxxxxxxxx:role/prefect-ecs-task-role",
).run()

If anyone could help me figure out what I could try to get this working, I would be really grateful.

Hi guys, as I couldn´t find that info somewhere else: What are the recommended minimal specs (CPU and RAM) for running a prefect server? I ask, because I am running an E2.1 Micro instance (Oracle always free tier) with one OCPU (2 threads) and 1GB of RAM. Thanks

Hey folks, I have a bit of an odd question, but I'd say it's more a matter of opinions: should Prefect be used to orchestrate everything ? Lemme explain what I mean with an example: I am refactoring my data stack and have some old pipelines in my AWS account that I need to carry over to the new stack. These pipelines are a mix of cloudwatch events, lambda functions, glue jobs, even step functions in one case. If moving to Prefect, how can I integrate these legacy pipelines with the new ones? Is it even possible to do so? For example, how can Prefect know when a certain Glue Job or Step Function finishes running or fails? Let me know what you think!

We have been getting tons of issues with our Prefect deployments without any changes to our deployments. They suddenly stopped running and all would go into "late" statuses for hours on end. When starting the agents locally, I see this error repeated endlessly. Any idea what may be causing this?

prefect.exceptions.PrefectHTTPStatusError: Client error '422 Unprocessable Entity' for url '<https://api.prefect.cloud/api/accounts/abc/workspaces/xyz/deployments/None>'
Response: {'exception_message': 'Invalid request received.', 'exception_detail': [{'loc': ['path', 'id'], 'msg': 'value is not a valid uuid', 'type': 'type_error.uuid'}], 'request_body': None}
For more information check: <https://httpstatuses.com/422>
16:00:39.141 | ERROR   | prefect.agent - Failed to get infrastructure for flow run 'abc'. Flow run cannot be cancelled.
Traceback (most recent call last):
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/prefect/agent.py", line 322, in cancel_run
    infrastructure = await self.get_infrastructure(flow_run)
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/prefect/agent.py", line 381, in get_infrastructure
    deployment = await self.client.read_deployment(flow_run.deployment_id)
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/prefect/client/orchestration.py", line 1504, in read_deployment
    response = await self._client.get(f"/deployments/{deployment_id}")
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/httpx/_client.py", line 1757, in get
    return await self.request(
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/httpx/_client.py", line 1533, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/prefect/client/base.py", line 253, in send
    response.raise_for_status()
  File "/Users/thet/SystemInternal/information-extraction-pipeline/venv/lib/python3.10/site-packages/prefect/client/base.py", line 130, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__

Hello! Is there a way to run ECSTask on a SPOT instance? thank you

Hi, We have a deployment that is scheduled using the RRULE

FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR;BYHOUR=18;BYMINUTE=15;WKST=MO

. However, it seems to be scheduled twice within the same minute as can be seen in the screenshot. What could be the reason for this and how can it be solved? Thank you kindly!

Hi there, Does anybody know how to get flow output if I run it using

run_deployment()

?

Hi, I'd like to get access to the

State

object for a failed task, but I'm not sure how to do that when the task throws an exception. Any advice?

try:
    state: State = trigger_sync.with_options(
        name=connection.name
    )(
        airbyte_server=airbyte_server,
        connection_id=connection.connection_id
    )
except AirbyteSyncJobFailed as e:
   // state is undefined

Hi, is it possible to schedule a deployment with parameters (and multiples times), like do

add(2+2)

at 10AM every days and

add(3+3)

at 11AM every days ?

Hello all, I would like to ask for your recommendations on structuring flows/subflows/tasks. We are working with dbt and Prefect and would like to define as many steps as possible as tasks, so that the radar displays the entire job as meaningfully as possible (without having to jump from flow to flow or subflow; instead, we want to see an overall picture). We run into the problem that dbt_cli() itself is also a task (correct me if I'm wrong). What are your recommendations here? How do you proceed with the interaction of Prefect and dbt? Many thanks in advance @Florian Pape, fyi

In v2 what are you using in place of state handlers? I would like to build a consistent way to send a customized slack notification when any task in a flow fails.

I have 3 flows for a data workflow (flow1=extracts the data, flow2= makes data ready, flow 3= creates training set) For the next step of work, we want to run this workflow (3 flows) in parallel for different parameters +1000 times . Our concerns are: 1- memory 2- how to set up the pipeline to run +1000 flows in parallel + run the subflows sequentially Any thoughts?

I have an oddball question: In a self-hosted Orion instance is there any possible way to lock blocks behind a secret token or something such that anyone with access to the UI can't create/update/delete (intentionally or accidentally)?

Hi All, is it possible to create workpools with Python API, if so please point to an example, thanks

Hello, facing a weird error when trying to deploy my flow with S3 as my storage:

botocore.exceptions.ClientError: An error occurred (MalformedXML) when calling the PutObject operation: The XML you provided was not well-formed or did not validate against our published schema

It worked fine yesterday. I’ve tried deploying using Python as well as CLI. Both give the above error. No permission changes on S3 and no changes on my Python packages. Any insights?

Is there any way to get flow run url from the flow ?

I want to have a stage in my .gitlab-ci.yml run every script with a Deployment.build_from_flow and apply. I did this in Prefect run with register and used wildcard for path. But in Prefect 2 I am finding it better to build and apply deployments through scripts. But I would also need that step automated in the ci pipeline. How can I do this? As a side note it seems you have to be in the scripts directory when you do this otherwise you get the entire directory structure below saved to your storage block as well…

Suppose I have a mapped task that returns a tuple... I want to further map out each element in the tuple's list to a downstream task. How do I do this? And how do I gather the results back such that I can associate them with the original tuple?

@task
def som_task():
  ...
  return name, list

@task
def mapped_process():
  ...
  return ??

@flow
def my_flow():
  names = ['test1', 'test2']
  x = some_task.map(names)
  # x = [('test1', [1,2,3,4,5]), ('test2', [6,7,8,9,10])]

  lots_of_data = mapped_process.map()

Hi community, i have a simple requirement to setup prefect with aws rds as db and SSL enabled. I am getting below error if i mount sever cert to prefect pod. Can someone help me with this? File “/usr/local/lib/python3.9/site-packages/asyncpg/connect_utils.py”, line 881, in _connect return await _connect_addr( File “/usr/local/lib/python3.9/site-packages/asyncpg/connect_utils.py”, line 768, in _connect_addr return await __connect_addr(params, timeout, False, *args) File “/usr/local/lib/python3.9/site-packages/asyncpg/connect_utils.py”, line 831, in __connect_addr await compat.wait_for(connected, timeout=timeout) File “/usr/local/lib/python3.9/site-packages/asyncpg/compat.py”, line 56, in wait_for return await asyncio.wait_for(fut, timeout) File “/usr/local/lib/python3.9/asyncio/tasks.py”, line 479, in wait_for return fut.result() asyncpg.exceptions.InvalidPasswordError: password authentication failed for user