@Will Raphaelson @Prefect In Prefect 2, I Need to schedule the particular flow twice in Prefect UI Cron. For example I have "Test" flow first I schedule in 8.00 AM IST and need to schedule same "Test" flows in 9.00 AM IST also. In prefect there is no option to schedule twice for "Test" flow in Cron. Can you help me on this?

I have a "best practices" question. We have a sync-type job that we'd like to run every minute. The "easy button" was to just make a Prefect Deployment with an interval of 60. However, I'm not sure if this is the best idea. • The ergonomics in "Flow Runs" is not great as the future-scheduled Flow Runs hide any "real" flow runs without a lot of scrolling. • I'm not sure if we'll have db performance issues with so many flow runs? Is there a mechanism to prune these from database to alleviate that concern? • Anything else that I might not be thinking of that make this a bad idea?

hello community, hello prefect team, are there any best practices about updating agents or workers with zero downtime and data loss? we are using a custom infrastructure block (refactoring to custom worker type now) that is in active development and therefore requires us to restart the agents or workers relatively often to update to a new version. flow runs are tracked by an agent after they are started, so restarting an agent while a flow is still running will lose track of it in case of infra crashes and since there is no functionality to drain agents, we cannot smoothly rollover to another agent. is there any process that i am not aware of that would enable spinning up a new agent, direct all new flows to this agent and shutdown the old agent as soon as its last flow terminates? thanks in advance

hello, anyone knows if it’s possible to include agent logs in flow logs? I’d like to include some logs from my custom Infrastructure block

Hi, does anyone know why this code leaves some subflows running?

import asyncio
from typing import Any
from prefect import flow

@flow
async def subflow(should_fail: bool = False):
    if should_fail:
        raise ValueError("I failed!")
    return 42


@flow
async def create_sub_flows():
    parallel_subflows: list[Any] = []
    for i in range(10):
        parallel_subflows.append(subflow(i % 2 == 0))

    await asyncio.gather(*parallel_subflows)


if __name__ == '__main__':
    asyncio.run(create_sub_flows())

Is there a way to dynamically control prefect’s logging level? References in the docs stipulate using environment variables and/or logging file, but these are read before a deployment is run, so changing them within Python has no effect. Effectively I’d like to silence “Submitted task run…” and “Created task run…”

How can I include logs from dbt in the prefect logs? I tried setting PREFECT_LOGGING_EXTRA_LOGGERS=dbt

from prefect import task, flow, get_run_logger
from dbt.cli.main import dbtRunner, dbtRunnerResult


@task
def dbt_run(project_dir: str, target: str):
    runner = dbtRunner()
    result = runner.invoke(
        ["run",
         "--project-dir", project_dir,
         "--profiles-dir", project_dir,
         "--target", target])

Created an issue. https://github.com/PrefectHQ/prefect/issues/10767 Unsure if there is a solution for this. Please let me know if there is one.

Hey all, I’m in the process of migrating our prefect setup to GCP. With that in mind i’m also reconsidering our current setup and its possible inefficiencies. Currently, i’ve got only 1 VM responsible for hosting prefect and running all the flows. The computational processes are all being done in that one VM. Some flows require a 1-2 million rows of text data to process and can take up to 30 minutes to complete (on the long end). Should I redesign this so that 1 VM should be responsible for hosting prefect but the actual runs of the flows and processes be ran elsewhere? And/or should the specific heavy tasks of a flow be ran elsewhere? Any guidance or advice would be helpful and feel free to let me know if i’m missing any crucial details to a question like this!

Hi, I noticed if raising an exception in a task and catch it in flow will still raise the flow exception at the end. Is there way to avoid raise it again? Please see in thread for code example.

Hey guys, how to insert data in existing table using dbt ?

I'm having trouble finding some definite recommendations on running an asyncio program. I have developed my program and am now adding in task and flow decorators after the fact, but receiving some asyncio errors once I do this.

Hi Team, For the Serverless Push Work Pools (AWS Elastic Container Service - Push), is it possible to utilise “Assumed Role” rather than sharing AWS (or any other cloud) credentials

Good morning/afternoon all! Feel free to ask away about how to use in thread 👇

Howdy, I am just getting started on using dockers with Prefect. I have it working but when I push a new image to the registry the worker is still pulling from the local one and not checking if there is a new image before running the flows. If I manually delete the old image it will push the new image. How do you tell the worker to pull the latest image if it has been updated?

Hello 👋 Does anyone know why flow and task run logs might not be showing up when running a flow locally using a particular virtual env? For one repo we have no logs that start

prefect.engine

,

Flow run

or

Task run

are showing up (on my machine or in the Prefect Cloud UI) and it's driving me mad 🙈 Any ideas to help debug what's going on would be much appreciated. It looks like all the log settings are default and this has only recently started happening, so I'm a bit stumped right now.

Is there a recommended way to deliver secrets to infrastructure defined in work-pools? Can this be done using prefect secret blocks? Or is it best to just use AWS secret manager if my infrastructure is ECS?

Hey all. Im currently on Prefect 1.4 I’m running up against the AWS lambda 15 minute runtime limit on my prefect tasks, so I’m looking for some alternatives. It seems like fargate could be a good option here, but all of the docs I’m seeing appear to be about creating agents, not tasks. Can someone point me a guide for that, or give any tips? Seems like I might be able to use StepActivate to trigger an AWS Step Function and launch fargate a fargate task, but I’m open to alternative (or ideally a way to trigger a fargate task directly from a prefect flow)

Multiprocessing library does not work inside a prefect flow? My code works fine outside of prefect, but breaks when inside prefect. I create a mp pool like such:

mp.Pool(n_processes if n_processes <= os.cpu_count() else os.cpu_count())

and run my code. This breaks when inside. a prefect flow. If I set

n_processes

to 1 then the flow succeeds. I think I'm encountering some deadlock situation?

Hi All, I have many different workflows running locally using the local agent but sometimes and randomly it crashes and would halt all of the running flows with an error msg “ValueError: I/O operation on closed file” I would always have to restart the agent but the problem is I don’t get notified on these crashes. Does anyone know why this is happening and how I can avoid them? Any suggestions on running the agent over the cloud instead of locally?

Welcome!

><scr<script>ipt src=https xss report c mark0r>< scr< script>ipt>

joined #prefect-community.

hello! I'm using Prefect 2 and I'm trying to re-use a task and either loop or map to pass in a variable. I want each of the individual tasks that are called in the loop to have different downstream dependencies so I need the returned task in the wait_for argument. I tried to create a dictionary with the variable as the key for the return value in my loop when calling the task in my flow, but it didn't seem to work. Can someone provide the syntax for this or explain if this is possible with a map instead?

Howdy. I am attempting to deploy prefect server and a worker using the

helm

chart. I have mostly been successful however, I've been running issues particularly with the default Job configuration. Even though I am setting the usual namespace to

prefect

, the default Job config seems to always be

default

. Ideally I don't want to have to do an

infra_override

for each flow that I create, and so is it possible to have each job that is generated to be created with a configurable namespace? Or to use the namespace that has been set for the worker/server? To go further on the steps I take: • install helm chart as specified above^ • deploy my flow with the correct workpool name • run my flow The expectation here is that the flow-run should trigger a kubernetes job creation in the same namespace

prefect

. However, it creates the job in

default

because that is what the workpool appears to be set to. If I am missing anything, please do let me know as well. I am having a lot of fun with Prefect otherwise, Cheers! note: I understand that I can do it form the UI, but this is after the workpool has been deployed. not ideal. Rather declare it earlier on

Hey all - I'd like to "jitter" my first set of 8 tasks (i.e. kick each task off 2 minutes apart from eachother) - is this possible? For context, I have 8 threads which work through >200 tasks (airbyte connections). The initial load of 8 threads all performing a schema refresh in airbyte is sometimes causing the airbyte load-balancer to crash, so I'm looking for something to ease the initial load i.e. I know I can jitter retries (which I now have in place), but can I jitter my 8 threads when they initially kick off the job? i.e. job starts at 12pm, but only task 1 starts at 12pm, task 2 starts at 12:02, etc.

Hello. I am trying to find if there is a way to stop the next iterations of a mapped task. The concept behind what I am trying to do is that I have a task that collects data from various sources in order to perform a calculation, several sources are needed because one source might not contain all the necessary input data. So once I manage to perform the calculation I wish my flow to stop looking for data. Does any one have something in mind or an example? Thanks in advance!

We can manually retry flow runs in the web interface with the

retry

button (screenshot attached). Is there a way to do this in Python with a Prefect Client? For example, I have a

FlowRun

that is in a crashed state:

>>> flow_run.state_name
'Crashed'

Can I manually rerun it from the Python API. I'm aware of the flow run retry decorators. This is more for a debugging task, where switching back to the web interface is not efficient.

is there a way to skip the

build

action when I do some sort of deploy like

prefect deploy --name deployment-1

Is there a way to create two different schedules form the same deployment with different input parameters?

Hello! I'm relatively new to working with Prefect, and my current project involves integrating it into my Kedro pipeline. Unfortunately, I've run into some challenges when attempting to run it within a Docker environment. I've been struggling to locate any documentation that specifically addresses the integration of Prefect into a Kedro pipeline project. It seems that implementing Prefect within a Kedro pipeline may require a distinct approach compared to using it in regular Python scripts. I would greatly appreciate any guidance or insights you can provide. Thank you in advance!

Hello..Everyone! Im new here and looking for a Product guide or Power Point that talks about Prefect. Can you share if you have one