Are there any links to CI/CD best practices with prefect v2 - I am transitioning from v1 and would like to keep things as close as I can - I currently bundle all my flows into a single container and run on GKE - I think I will probably move to using cloud run but would like to keep the general pattern of merging to main, building a container and then thats the container which runs the flow?

When building a flow of flows in Prefect 2, does anyone know how you might suppress notifications sent via

on_completion

and

on_failure

that would be send via the subflows? (or if it's possible at all?) There are some cases where the subflow is itself useful, but we've incorporated it into a flow of flows as well so we would want notifications from both places but not in both situations

We've been migrating a flow from v1 to v2 and that's all working out fine. However, in v1 we would invoke a flow from Python using code a little like this -

from prefect.client.client import Client

c = Client(api_token='xxxxxxxxxx')
params = {'param1': 'Foo', 'param2': 'Bar'}
c.create_flow_run(
  version_group_id='xxxx-xxxxxx-xxxxxx-xxxxx', 
  run_name='my_run_name',
  parameters = params)

But I can't see a good example of how to do the same with v2! I think I'm missing something fundamental! Thanks

If i’m updating the code of a deployed flow, do i need to delete the flow and redploy? Or will runs reflect the changed code automatically?

hi, looking into 2.10.5, is there a way to go from flow run to flow? in 2.10.4 there was a link right above the Flow Run section but I don’t see that link anymore in 2.10.5 in the header or the Details tab

Heya! How do I disable the limit on the number of Runs that get scheduled? Using 1.3.1. Registering with the

prefect register

CLI command, using the

ECSRun

Run Config. The schedule is defined in the

.py

file and passed as an argument to the flow (using a list of

CronClock

objects passed to a

Schedule

object).

Any ideas as to what might be causing this h2 error? My flow and many tasks run successfully. Then, the following task seems to produce the failure (both in thread). The task itself is mapped with a concurrency limit of 8. Thousands of api calls are being made. I have not found any other error messages in the logs so far, but I'm continuing to look.

With the upcoming Legacy Prefect Cloud 1 account freeze, if I upgrade my free tier account now to a “starter” counterpart will this delay the freeze to 15/07?

hello community, hello prefect team, we were recently developing a new flow with very granular tasks which eventually resulted in a flow with 50k+ task runs. since we have no blocking statements like wait or result in our flow, this causes the flow to submit all these tasks at the very beginning. Since the PrefectClient has a connection pool of only 16 connections and a pool timeout of 30s, this led to a large number of pool timeout errors. increasing the timeout could prevent the timeout errors but submission of tasks still took very long and it seemed like the agent was only occupied submitting tasks and started a way lower number of tasks at a time than expected. We eventually refactored our flow to use less granular tasks and parallelize over multiple flow runs using a flow of deployments in order to decrease the number of tasks per flow. So i was wondering if you have any recommendations or experience with large number of tasks. how much can we expect prefect to handle? what's a sustainable task submission rate that wouldn't overwhelm the prefect client and are there any special best practices for massive flows? Thanks a lot.

Hey folks, is it possible when running a script with prefect, to specify which flow to run? E.g. imagine this:

@flow
def first_flow():
    ...

@flow
def second_flow():
    ...

def main():
    first_flow()
    second_flow()

What if in certain situations, i want to run only the 2nd flow? Is that possible / reasonable ?

Hi All, we are migrating from v1 to v2, we wanted to implement something like below, but if sub flow fails parent flow also fails but wanted to run clean up flow regardless of upstream flows fail state

from prefect import task, flow, get_run_logger, allow_failure
from prefect.task_runners import SequentialTaskRunner

@flow
def flow_1(p: int) -> int:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f" \n\n ----------------Flow 1 --{p}-------------\n\n")
    return 5

@flow
def flow_2(m: int):
    raise ValueError("Non-deterministic error has occured.")


@flow
def combine_data(p: list) -> int:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"-\n\n------------Combining data -- {p}-------------\n\n")
    return sum(p)

@flow
def additional_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"-\n\n------------additional_flow -- {p}-------------\n\n")
@flow
def clean_up_task(result: int):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"\n\nCleaning up 🧹 --- {result}-----\n\n")
    <http://logger.info|logger.info>("\n\nCleaning ------------------------------ up 🧹\n\n")


@flow(task_runner=SequentialTaskRunner)
def allow_flaky_transformation_to_pass():
    f1 = flow_1(1)
    f2 = flow_2(2)
    result = combine_data.submit(
        [f1, f2], wait_for=[allow_failure(f1), allow_failure(f2)]) # wait for all upstream flows
    additional_flow()
    clean_up_task.submit()  # This should always run even if all of the above flow fails


if __name__ == "__main__":
    allow_flaky_transformation_to_pass()

Hi, Facing an issue in launching agent in the Kubernetes deployment. I have created the work pool and work queue in my prefect UI. Able to launch agent in my local machine by pointing to the a work pool and work queue the agent is getting started perfectly.

prefect agent start -p <WORK_POOL> --api <WORK_SPACE_API>

But when i try to do the same in the Kubernetes deployment. It is trying to create a new work queue even if the work queue is already present in the work pool and getting failed.

This message was deleted.

Hi everyone! I have a flow which I want to deploy multiple times with different parameters. The only problem is that I want some parameters to be not modifiable when a user does a custom run. How can I achieve this?

Hello everyone, We have been using Prefect 1 with a local Docker agent, but we are now trying to migrate to Prefect 2. However, I have noticed that there are many changes in syntax and conception. I was wondering if anyone knows whether it is possible to migrate all of our flows at once, or if we need to modify them individually. Thanks a lot

I’m executing

get_run_logger().info(get_client().read_deployment(FlowRunContext.get().flow_run.deployment_id).name)

in a flow and I get

getAttributeError: 'coroutine' object has no attribute 'name'

I think this is something related to async coroutines. How can I solve it?

Is it possible to use multiple storage blocks in a deployment? We have individual github repositories for each unit operation in a facility; however, we’d like to refactor some of the redundant code into a common repo. Is it possible to create a deployment that references 2 github storage blocks?

We're running Prefect Orion in a self-hosted Kubernetes cluster. We upgraded from 2.7.4 to 2.8.0, and now when we run any flow, we immediately run into an error where Prefect can't load its state.

2023-04-21T15:16:04.920982508Z /usr/local/lib/python3.8/runpy.py:127: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; thi
s may result in unpredictable behaviour
2023-04-21T15:16:04.921075059Z   warn(RuntimeWarning(msg))
2023-04-21T15:16:04.997977913Z 15:16:04.996 | INFO    | Flow run 'phi321-lysenko-tensor' - Created task run 'generate_operation_id-a9e17ed9-0' for task 'generate_operation_id'
2023-04-21T15:16:04.999103995Z 15:16:04.997 | INFO    | Flow run 'phi321-lysenko-tensor' - Executing 'generate_operation_id-a9e17ed9-0' immediately...
2023-04-21T15:16:05.043778806Z 15:16:05.042 | INFO    | Task run 'generate_operation_id-a9e17ed9-0' - Task run '43baff18-8292-4cd7-a694-2dcf6d1a0c77' already finished.
2023-04-21T15:16:05.119806977Z 15:16:05.044 | ERROR   | Flow run 'phi321-lysenko-tensor' - Encountered exception during execution:
2023-04-21T15:16:05.119866174Z Traceback (most recent call last):
2023-04-21T15:16:05.119875789Z   File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 637, in orchestrate_flow_run
2023-04-21T15:16:05.119882275Z     result = await run_sync(flow_call)
2023-04-21T15:16:05.119888470Z   File "/usr/local/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
2023-04-21T15:16:05.119896590Z     return await anyio.to_thread.run_sync(
2023-04-21T15:16:05.119904639Z   File "/usr/local/lib/python3.8/site-packages/anyio/to_thread.py", line 31, in run_sync
2023-04-21T15:16:05.119911921Z     return await get_asynclib().run_sync_in_worker_thread(
2023-04-21T15:16:05.119917612Z   File "/usr/local/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
2023-04-21T15:16:05.119924062Z     return await future
 ...etc...
2023-04-21T15:16:05.120058513Z   File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
2023-04-21T15:16:05.120063946Z     raise self._exception
2023-04-21T15:16:05.120069588Z   File "/usr/local/lib/python3.8/site-packages/prefect/engine.py", line 1068, in get_task_call_return_value
2023-04-21T15:16:05.120078014Z     return await future._result()
2023-04-21T15:16:05.120085086Z   File "/usr/local/lib/python3.8/site-packages/prefect/futures.py", line 237, in _result
2023-04-21T15:16:05.120091377Z     return await final_state.result(raise_on_failure=raise_on_failure, fetch=True)
2023-04-21T15:16:05.120097177Z   File "/usr/local/lib/python3.8/site-packages/prefect/states.py", line 100, in _get_state_result
2023-04-21T15:16:05.120103360Z     raise MissingResult(
2023-04-21T15:16:05.120110136Z prefect.exceptions.MissingResult: State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.
2023-04-21T15:16:05.141538481Z 15:16:05.140 | INFO    | prefect.engine - Engine execution of flow run '5c996762-8edc-4584-9f70-4d1d6adf2b3b' aborted by orchestrator: This run has already terminated.

Hey there - I’m jumping back into using Prefect again after a while and I’m trying to bring myself up to speed with all the changes. I did a bit of work with 2.x about 6-9 months ago - but looks like there’s been a fair amount of change with work-pools and workers. I’m getting a little confused about the relationship between agents and workers - are workers essentially a replacement for agents? or are they supposed to work together?

Are

on_completion

and

on_failure

hooks ignored when a task state is

Cached

? That seems to be the case, but AFAIK there’s no docs (https://github.com/PrefectHQ/prefect/issues/8967)

#prefect2.10.5 prefect UI is not displaying flow runs on Windows, If I access the same UI on linux I see the flow runs.

Hello, consider I have a prefect server on EC2 (AWS), can I deploy a flow in a sagemaker processing job and see the flow in the EC2 server?

Hey Prefect community ! Quick question that I have difficulty to find with the document. I am currently creating deployment through the CLI like :

prefect deployments build \
    my_flow.py:my_func_flow \
    --name super-cool-deployment \
    --path /usr/src/app \
    --work-queue test \
    --skip-upload \
    --apply

I would like to add 30m interval with max 15 runs. Can I do it through terminal ? Or do I need to use a yml file ? Thanks in advance community !

Hi guys, I am trying to migrate all of my prefect v1 code to v2. I was doing this in v1 :

prefect_client.create_flow_run(flow_id=pf['flow_id_prefect'], parameters=pf["parameters"], run_name=pf["run_name"],idempotency_key=run_key)

Now, I updated the logic with v2 deployment and deployed it using .yaml file. I could see the deployment in

prefect deployment ls

and could trigger it from cli using

prefect deployment run --params

but I want to trigger it using the python api. I found

run_deployment

method

from prefect.deployments import run_deployment

but this wait till all the process finishes which is not what I want. I dont want polling. Please do suggest, How I could move forward with it ?

Hi everyone, I have multiple flows deployed on schedule which runs around 100 sub flows each. This is causing a massive clutter in the UI. I want to be able to have stronger filters in the UI with tags. But it isnt very clear on how I can create tags for my deployed flows and generated sub flows. I have created an issue in github https://github.com/PrefectHQ/prefect/issues/9078 . Please let me know if you have any solutions to this.

Hi everyone, short question: Why is it that when we use a remote infrastructure (like Docker, Kubernetes Job), we need to provide a remote storage, is it an antipattern to store the code in the custom docker image?

Hi there! We want to programmatically trigger flow runs from deployments. To do that, I’d like to confirm that is the deployment_id immutable, or should we read the deployments list first to get the deployment_id every time?

Hi all, sharing a post I made today on how I deployed a worker on kubernetes. I've been querying this channel a lot recently, it wraps up all I learned from the support I got to deploy my worker and hopefully will save some time for others too! https://discourse.prefect.io/t/running-a-worker-as-a-kubernetes-pod/2821

I'm trying to define correctly type-annoted flows, but, e.g.

val = tasks.my_task.with_options(timeout_seconds=600)()
val2 = tasks.my_2nd_task(val)

It seems that

.with_options

,

.submit

etc. do not define their return types. Prefect version is 2.9.0.

Hi guys,

from prefect.deployments import run_deployment

I have a weird problem, when I use the line of code, logger in my script only works for warning and error not for info and debug. any hints how to solve this ?