has anyone ever ran into their flows running fine on local, deploys fine to prefect cloud and each task runs fine but the flow crashes on the final step? each subtask showing completed, flow is failed, no data gets written to our DB in the final step of the flow (even though it shows completed). I'm struggling to debug

At runtime how do we access artifacts stored in S3 block storage? (Prefect v2) Are all artifacts copied to some location when the flow run begins?

Hi there! I have a flow which has been failing at an increasing frequency recently. The actual task logic is always completed, but then I get

Finished in state Failed('Flow run encountered an exception. MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.\n')

at the end. Manually retrying the flow succeeds, but I do have

retries=3

in the

flow

decorator and the flow doesn't appear to be retried

This message was deleted.

I am using the

create_flow_run_from_deployment

/

run_deployment

APIs to compose my deployments on http request. E.g. I get a payload defining a set of flows to chain together, that triggers a parent container flow, and that container flow does the orchestration of tying things together. The way I'm thinking of running it, because there are dependencies, is to have the

run_deployment

code wrapped in a task like this :

@task
async def run_deployment(depl_id: str, parameters: dict):
    async with prefect.context(**prefect.context.run_params):
        async with prefect.Client() as client:
            run = await client.create_flow_run_from_deployment(deployment_id=depl_id, parameters=parameters)
            run_state = await run.get_state()
            return run_state.result

and then in an orchestration flow I think I want to do something like this where I create these tasks for each deployment.

@flow
async def container_flow(flow_graph: dict):
    results = {}

    ordered_flows = order(flow_graph)

    # create tasks
    tasks = {name: run_deployment.map(deployment_id=flow_params['deployment_id'], parameters=flow_params['inputs'])
             for name, flow_params in flow_graph.items()}

    # set dependencies
    for flow_name, flow_params in flow_graph.items():
        for dependency in flow_params.get('dependencies', []):
            tasks[flow_name].set_upstream(tasks[dependency], flow=True)

    # run tasks concurrently
    flow_results = await prefect.engine.run(tasks, return_tasks=True)

    # store results
    for flow_name, flow_result in flow_results.items():
        results[flow_name] = flow_result.result
    return results

I am trying to use the

run_deployment

orchestrator pattern to chain together deployments dynamically. Some of these deployments depend on others. I've attempted to finagle a way around this by wrapping

run_deployment

in a task as such:

@task
async def run_deployment(depl_id: str, parameters: dict):
    async with prefect.context(**prefect.context.run_params):
        async with prefect.Client() as client:
            run = await client.create_flow_run_from_deployment(deployment_id=depl_id, parameters=parameters)
            run_state = await run.get_state()
            return run_state.result

I then create tasks as such

tasks = {name: run_deployment.map(deployment_id=flow_params['deployment_id'], parameters=flow_params['inputs'])
             for name, flow_params in graph.items()}

and set their dependencies:

# set dependencies
    for flow_name, flow_params in graph.items():
        for dependency in flow_params.get('dependencies', []):
            tasks[flow_name].set_upstream(tasks[dependency], flow=True)

the goal here is to dogfood the taskrunner to chain these together and kick off the flow with something like

await task_runner.submit(tasks)

Two issues I see with this though: 1. Some of the

parameters

to downstream

create_deployment

runs are derived from upstream deployment runs. I wonder if anyone knows an elegant way of doing that? 2.

await task_runner.submit(tasks)

would actually work as anticipated?

Hi experts~! I trying to migration prefect 1.x to orion and somebody already are migrated. Let me introduce my condition ; • successfully set the secret block at orion server and use the secret by using get method untill last week • the secret contains the database connection information • last weekend, our office had been fixed the whole wetwork • then edit the secret block's value (changed connecting IP)

Hi experts~! I trying to migration prefect 1.x to 2.x orion and a issue after office network re-construction. My conditions ; • prefect version == 2.6.7 • Untill last week successfully set the secret block at orion server and use that by using .get method. • The secret contains the database connection information. • Last weekend, our office had been fixed the whole wetwork. • Then i edited the secret block's('3f') value as changed the new IP address.

RuntimeWarning: coroutine 'Block.load' was never awaited
  secret_block = Secret.load('3f')
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

• So I make a new secret block and that's also don't working with below error message..

C:\Users\user\AppData\Local\Temp\ipykernel_12716\3323813457.py:3: RuntimeWarning: coroutine 'Block.load' was never awaited
  secret_block = Secret.load("new-net")
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In [27], line 6
      3 secret_block = Secret.load("new-net")
      5 # Access the stored secret
----> 6 secret_block.get()

AttributeError: 'coroutine' object has no attribute 'get'

Is there are happen to trouble by new network? Thanks for your reply.

Is there a way to rename a flow already registered with Orion/Cloud? I want to change its name and not lose run history

Howdy. Is it possible to invoke Glue Job using Prefect?

Hi. Can you explain what does it mean, it happens on retry?

State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.

I have issue with Prefect Agent Does any one have solutions? Thank you 🙌 Also, it seem like the error happens in this path :"C:\Users\Will Truong", but i don't know what the script want to open/run when access this path.

Microsoft Windows [Version 10.0.22621.963]
(c) Microsoft Corporation. All rights reserved.

D:\Athena\etl>prefect agent start "work_queue_for_test"
Agents now support multiple work queues. Instead of passing a single argument, provide work queue names with the `-q` or`--work-queue` flag: `prefect agent start -q work_queue_for_test`

Starting v2.7.4 agent with ephemeral API...

  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from queue(s): work_queue_for_test...
16:18:20.487 | INFO    | prefect.agent - Submitting flow run 'fd08f02a-18ca-4538-bdf1-728dc7e88e22'
16:18:20.604 | INFO    | prefect.infrastructure.process - Opening process 'beta736-bajor'...
'C:\Users\Will' is not recognized as an internal or external command,
operable program or batch file.
16:18:20.622 | ERROR   | prefect.infrastructure.process - Process 'beta736-bajor' exited with status code: 1
16:18:20.651 | INFO    | prefect.agent - Completed submission of flow run 'fd08f02a-18ca-4538-bdf1-728dc7e88e22'
16:18:20.668 | INFO    | prefect.agent - Reported flow run 'fd08f02a-18ca-4538-bdf1-728dc7e88e22' as crashed: Flow run infrastructure exited with non-zero status code 1.

Hello People, I am looking a way to create pagerduty notification via prefect python api. Right now the only way I can create this notification is via Prefect Automations using Prefect UI. Any docs on this would be really appreciated. More context: I have created PagerDuty integration using prefect actions but I want to use this action pragmatically and send the notification to PagerDuty rather than creating prefect automation via prefect UI.

Hey prefect people! Can you tell me please is it possible to specify a schema while connecting prefect to Postgres database?

Hi, Is there a difference between task-runs of a sub-flow and task-runs of a stand-alone flow? I observe that a flow-run does not use cached results if the cached results were created by the same flow executed as sub-flow.

Hi, we're trying to setup Prefect1 with GreatExpectation and PostgreSQL data backend and Dask flow executor. Has anyone tried similar setup, because we're facing some challenges?

Hi all! Happy New Year! Question: are deployments in Prefect2 the same as registering in Prefect1?

Hi all, quick question on the

prefect.deployments.run_deployment()

function: If I deploy a flow without specifying a deployment name, how can I use

run_deployment()

to trigger a flow run? Would it just be

run_deployment(name='flow_name')

? related docs: https://docs.prefect.io/concepts/deployments/#create-a-flow-run-in-a-python-script

Hey! Question about flow cancellation & Dask workers on k8s (Prefect 2.7.7) I’ve got a Kubernetes deployed flow, which uses dask_kubernetes and spawns dask pods (scheduler + workers) to execute the tasks. When I cancel a running flow from the cloud UI, it terminates the flow run and deletes the prefect job pod, but it leaves the dask pods there just hanging - still alive, but not actually doing anything. Is that intended, and if so, any suggestions on best approach for auto-cleanup of these? At the end of a failed or successful flow run these are auto-terminated, but not when the flow is cancelled. Thanks!

hey all, in prefect 1, how do i exit a flow if a task fails? when my task fails, the flow continues to run tasks that are not dependent on that task.

What is the recommended means by which one can send a slack notification from inside of a task? It looks like the prefect-slack task does not work if called as a function?

From reading other messages I am aware the statefulness classes does not work well with class method as tasks or flows. I have however written my python code using classes, I like them for keeping my code in manageable segments. As long as I do not need to pass any self attributes from when task to another I assume I am ok to use a class? I am thinking I have a separate

prefect-flow.py

file that imports my class. Should I create an instance in the flow and pass to each task? Or create a class instance in each task?

Question: did prefect2 get rid of prefect.config?

Hi all - we’re seeing sporadic

BrokenPipeError: [Errno 32] Broken pipe

crashes in one of our flows on 2.7.7, running on DaskTaskRunner. This flow runs ~1000 tasks, occasionally one of these tasks will enter a Crashed state with this error and cause our flow to enter a Failed state. Retries on these crashed tasks don’t seem to work (I’m guessing Crashed state tasks are excluded from retry logic). Full traceback in the thread. Any ideas? Thank you!

Just a reminder, our usually-friendly bot Marvin is here to help answer any questions! Isn’t that right @Marvin?

Is there a way to change the location of the temporary working directory that is used by prefect v2?

Hi guys 👋 Someone have experience with the DaskTaskRunner() and runing tasks in parallel? We can’t make tasks run in parallel. I’ve detailed the problem here: https://discourse.prefect.io/t/tasks-are-not-running-in-parallel-when-using-dasktaskrunner/2180/1

Hi everyone, can someone enlighten this behaviour to me? I have the

case

with flow when is

true

and when is

false

But i noticed that when the case is

false

the

task_b()

is always SKIPPED. After some struggling 🥹 i figure out that the reason that the task was being SKIPPED because inside the case

true

i'm changing the

task_result

variable (if i rename the task run on false cond). Is this the desired behaviour? Why is like that? Is there a way to bypass without renaming?

from prefect import task, Flow, case


@task()
def task_a():
    print("task_a")
    return "task_a"


@task()
def task_condition():
    return False


@task()
def task_b(value: str):
    print("task_b")
    return f"task_b={value}"


@task()
def task_c():
    print("task_c")


with Flow("test-flow") as flow:
    task_result = task_a()
    cond = task_condition()

    with case(cond, True):
        task_result = task_b(task_result)
        task_c()

    with case(cond, False):
        task_result = task_b(task_result)

flow.run()

Hi all - we’re seeing sporadic

BrokenPipeError: [Errno 32]

Broken pipe crashes in one of our flows on 2.7.7, running on DaskTaskRunner. This flow runs ~1000 tasks, occasionally one of them will enter a Crashed state with this error and cause our flow to enter a Failed state. Retries on these crashed tasks don’t seem to work (I’m guessing Crashed state tasks are excluded from retry logic). Full traceback in the thread. Any ideas? Thank you!

@Marvin give us a joke about data engineering