This doesn't happens if I limit the number of tasks that have to be executed (I am using DaskTaskRunner)

hello! is there a way to retrieve a parameter value from a function call

on_failure

function handler either from the

flow

variable or the

state

variable are passed to that function?

I have tried a different task runnuer

same issue if I am suing CocncurentTaskRunner

it just hangs

This message was deleted.

Hey everyone, I have a question regarding scheduling of flows run, 1 have one agent running at moment, and i used the:

create_flow_run(flow_id=flow_id, parameters=parameters)

But sometimes when theres nothing to run or running, the flow is scheduled to run in like 20~30 min, even when i'm not providing the scheduled_start_time parameter, how can i make my flows always get reschedule immediately? (I was think that this is the default behaviour), maybe is the amount of agent that i'm running? Thanks.

I have a flow running with multiple functions which are defined in a custom block. Is there any way to create upstream and downstream dependencies between these functions. Since these are not tasks I can't use wait_for attribute. Any hints or pattern to handle this use case. Orchestrate different functions built in a custom block?

👋 How can I "batch" my pipeline? I would like to run the full pipeline in batches of ~10k rather than having massive updates where one failure would stop everything. And so I would like to have batches of get_user_ids() run everything in parallel (LocalDaskExecutor), save the results in Redshift, and start again. PS: Another approach would be to build and "incremental" logic from the source data PS2: I am using Prefect 1.0

Hello. Any plans in the future to have a terraform provider to manage organization resources? Things like workspaces, api keys, and things like that ? It would be pretty useful for enterprise that needs to manage their infrastructure with a fully automated IaC.

I am testing rerunning failed flow (Prefect 2.0) and get this message: prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

Streaming now youtube

Hello, for ECSTask, how can can I set up the block to use a specific task definition revision, rather than creating a new one every time? I'm running into a lot of

RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family.

errors

Hi Prefect Folks - The documentation for artifacts indicates it supports

Github-flavored Markdown

, but I don’t think

<details>

are actually supported. I searched for existing GitHub issues, but wasn’t able to find anything.

Has anyone notice a recent change is behavor of the

PREFECT_LOGGING_EXTRA_LOGGERS

mechanism? I am running a DaskTaskExecutor with a

SLURMCluster

backend to create the set of

dask-workers

. I can see that the loggers in modules I am using in my prefect2 pipeline are beingreported in the

slurm

stderr output that is written to disk through the

sbatch --error log.err

argument. These messages are following the prefect log handler configurtion ("Time | State | _ name _ - Message"), but they are not being saved to the orion database / presented by the orion UI.

Hi. What is the replacement for flatten in Prefect 2?

Is there a version of this diagram for Prefect V2?

Would really appreciate some help with this 😁: https://prefect-community.slack.com/archives/CL09KU1K7/p1669371891611149

My general pipeline flow is "fetch data" -> do various operations on data -> persist results. My first instinct was to split those stages up into separate (potentially mapped) tasks, but I'm worried about the overhead of passing large data blobs between tasks (especially if I move to running things on a cluster of some kind in future, where those tasks could end up running on different machines). Am I right to be worried about that? Is there a best practice?

Hi. Is it correct statement that Prefect 2 map function does not guarantee the order? Example:

l1 = task1.map([1, 2, 3])
l2 = task2.map(l1)

task3.map(l1=l1, l2=l2)

Prefect does not guarantee that in

task3

list

l1

elements will be passed with corresponding

l2

elements?

Teaser: look what just got merged 🎉 starting from the next Prefect release, you'll be able to do:

@flow(log_prints=True)

Hi there, I’m experiencing the same problem as this previous post. When testing a slack incoming web hook, the test fails with no error message. What would be the best way to further triage this issue?

Hello, I have an agent set up on ECS and built a deployment using the ECStask block to run the flow as a Fargate task. I tried to run the deployment and the flow run failed. I have no information as to why though. The logs are completely empty, it says "This run didn't generate Logs". There is also no logs in AWS for the task I was intending to run so it seems like it failed before the task even started.

Hey folks, my teammates and I are trying to migrate a monorepo of tasks & flows from Prefect 1 to Prefect 2. As part of that, we're naturally doing some testing. We were previously using the

is_successful()

method in an assert in our testing, and to move that to Prefect 2, we were thinking we might call

get_run_context()

to get the flow run state and return some value (likely 'Completed' from the

state_name

parameter). We've observed that calling

get_run_context()

within a flow's context causes the flow to hang. Has anyone observed this, or does anyone have any suggestions for remedy?

Hey prefect when creating a flow which would run in kubernetes. v1 adds automatically flow_id to the pod labels. But V2 doesn’t appear to so. How would I add flow meta data to the flow pod ?

Following up on Kevin's question above, what is the equivalent for the

is_successful()

method from Prefect 1 for Prefect 2? In other words, is testing the state name = 'Completed' the best way to check for success?

myflow_state = my_flow(return_state=True)

# Test if state is successfull:
if not myflow_state.name == 'Completed':
    raise Exception("myflow failed")

Hey all - Has anybody used

prefect-airbyte

to trigger syncs on Airbyte Cloud? Or is it only for open source hosted instances?

Hi! I'm running a flow that run two subflows. One subflow execute a task that uses

run_deployment

for execute two deployments that has to be completed the next subflow. I'm having issues in the execution of the seconds subflows, the agent dies after one hour of waiting. How can I apply this logic successfully? I need to execute n deployments and when the n deployments has been completed, execute another flow. Image 1: The deployments runs properly. Image 2: The tasks that runs the deployments never ends.

Hi all~ I am trying to migrate to the prefect 2.0 version!! marvin In prefect 1.0, the flow version was well displayed as a number in the upper right corner of the UI(below image, left), but does the orion UI only show the flow version id for each flow? (below image, right) I thought it was much more intuitive to show it in numbers like prefect 1.0. Is it i can't find it at the Orion? Or was it intended what just show flow version id?

Hi guy, I am running a agent from my ec2 machine , and after a while, the agent throws this error

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/prefect/agent.py", line 154, in get_and_submit_flow_runs
    queue_runs = await self.client.get_runs_in_work_queue(
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/prefect/client/orion.py", line 763, in get_runs_in_work_queue
    response = await <http://self._client.post|self._client.post>(
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1848, in post
    return await self.request(
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1533, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/prefect/client/base.py", line 160, in send
    await super().send(*args, **kwargs)
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1620, in send
    response = await self._send_handling_auth(
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1648, in _send_handling_auth
File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1685, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_client.py", line 1722, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_transports/default.py", line 352, in handle_async_request
    with map_httpcore_exceptions():
  File "/usr/lib/python3.10/contextlib.py", line 153, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/home/ubuntu/5c-ed-sumary-telegram-bot/env/lib/python3.10/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state ConnectionState.CLOSED

Any idea why this is happening ? I have deployed lot of deployments before, and never had such a issue