What is the best practice for the following situation? I have tens of thousands of mapped tasks. Suppose that each writes to a remote database. If that database goes down, I will see cascading failures. What I would like to do is a) manage the threshold of allowed failures across the aggregate mapped tasks b) when that threshold is reached, halt the workflow c) Be able to resume the workflow from the point of failure c) would ostensibly require caching behavior, I would think. But since there are no dependencies between the mapped tasks, I don't want to just cache the results of the successful tasks -- I just want to somehow remember the mapped tasks that already got ran to successful completion, and then, when the flow is restarted / resumed, only map across those tasks that did not run to completion / were not started.

hi community! is there a good way to invalidate a checkpoint? tried setting

cache_validator=prefect.engine.cache_validator.never_use

on the task but that doesn't seem to do the trick (running prefect core locally)

Does Orion support dynamic task naming yet? I tried the

@task(name="{val}")

syntax, but it didn’t look like the templating syntax was getting interpreted. (I would up with “{val}” in all of my task names)

Is it okay to return a fitted

scikit-learn

object from a task? Any gotchas if I then pass it to a second task which is responsible for calling

.predict()

using that same object. Reason I’m asking: I thought I read somewhere in the prefect docs to be careful with state in such cases, but I can’t seem to find it.

Hi! I have a flow using github storage that is scheduled to run (using prefect cloud). If I push new code and register it with, are the currently already scheduled runs going to use the new code when triggered?

prefect register --project $PREFECT_PROJECT_NAME --path flows/ --label prod --no-schedule

1. Hi all, we have been using Prefect for scheduling and monitoring flows as part of our organization work. We have a set of questions and issues we need answers for. Please address them if possible. We have a Docker Agent managing flows built with Docker Run as run config and Local Storage but the agent seems to stop picking and running flows after a few days - typically 6-7 days and even if it is actively polling, the flows are not run. Is this issue seem before and if so ,what is the solution? We start the docker agent with nohup prefect agent docker start --token <auth_token> -l <agent_label> 2. We have a flow which tries to establish a pyodbc connection to ms sql server with connection timeout of 60 seconds and have the task configured with max_reties and retry delay. max_retries=2, retry_delay=timedelta(minutes=1) We expect the task to finish running in under 10mins but they are running off into over 15-16hours. Why is this the case? This is also causing cannot allocate memory issue. 3. We have seen some scenarios like a prefect agent is down suddenly and this might lead to flows failing continuously. How can cases like this be avoided in future? For example : There are some scenarios where an agent is going down, at this time all the scheduled flows pile up and are tagged under Late runs. After few hours when the agent is up again, all the late run flows starts to run and this leads to a memory issue in the machine. 4. Is there a way in prefect, to kill a flow if its is running for more than 'x' amount of time. 5. Is there way in prefect to stop a pipeline schedule based on condition (eg: if the last three consecutive flow fails then stop the schedule).

Hey all, two quick questions 1. Is there a channel specifically for Orion questions? 2. Is there a way to specify a Dockerfile to use rather than an image for DockerFlowRunner? Context is basically I make heavy usage of things like

from mypackage.utils import frequently_used_task

where my flow is also inside of

mypackage

. Running outside of a docker container with

mypackage

installed means I hit an import error immediately.

Hey guys, I’m making a database query as a snowflake task inside a Prefect Cloud flow. The output of this query takes up memory size that’s larger than 1 MB so when the Task tries to call

set_task_result()

it fails with error message:

'State payload is too large.'

which makes sense. My Prefect infrastructure is sitting on a KubernetesJobEnvironment. My understanding is that to store this large result correctly I should follow the documentation here: https://docs.prefect.io/core/concepts/results.html#result-objects Are there any additional considerations I should take into account with setting up correct result storage with this infrastructure?

Hello, a question about cache. Is there a scheduled task or function that deletes expired cache?

Hi, could someone please elaborate what "Prefect init job" actually refers to here https://docs.prefect.io/api/latest/cli/agent.html#kubernetes-install? Are those cpu/mem resource options used for controlling the agent or the jobs?

I wrote a naive flow that looks like:

@task
def get_table() -> object:
  ... query table
  result: pandas.DataFrame = cursor.fetch_dataframe()
  return result

def load_to_s3(result: object) -> None:
  awswrangler.s3.to_csv(
    df=result,
    path='<s3://bucket/folder/table.csv>
  )

with Flow(...) as flow:
  get_data = get_table()
  save_data = load_to_s3(get_data)
  ...

I’d like to now extend this to somehow pass a list of tables from the same database to extract and load to s3. Is this possible given this simple template?

I'm new to Prefect and I'm having difficulty using/understanding arguments with the

DbtShellTask

. Here's some mock code:

from prefect.tasks.dbt.dbt import DbtShellTask

dbt_shell_task = DbtShellTask() 

with Flow("nightly_dbt_flow") as flow:
    env = Parameter('env', default='dev')
    run_task = dbt_shell_task(command='dbt run',
                              environment='dev')

flow.run()

I receive a TypeError that the keyword argument

environment

doesn't exist:

Traceback (most recent call last):
  File "nightly_dbt_flow.py", line 50, in <module>
    environment='dev')
  File "/Users/daniel/git-repos/bi-prefect/.python/lib/python3.7/site-packages/prefect/core/task.py", line 662, in __call__
    *args, mapped=mapped, upstream_tasks=upstream_tasks, flow=flow, **kwargs
  File "/Users/daniel/git-repos/bi-prefect/.python/lib/python3.7/site-packages/prefect/core/task.py", line 702, in bind
    callargs = dict(signature.bind(*args, **kwargs).arguments)  # type: Dict
  File "/Users/daniel/.pyenv/versions/3.7.10/lib/python3.7/inspect.py", line 3015, in bind
    return args[0]._bind(args[1:], kwargs)
  File "/Users/daniel/.pyenv/versions/3.7.10/lib/python3.7/inspect.py", line 3006, in _bind
    arg=next(iter(kwargs))))
TypeError: got an unexpected keyword argument 'environment'

Why would

environment

cause an error but not

command

? When I remove

environment

the task does get attempted without a

TypeError

.

environment

and

command

are both arguments defined in the

DbtShellTask

class. I'm on prefect version

1.2.0

.

We’re getting into operations using Prefect and wondering if there is a way to run a re-run a failed (or perhaps not failed) subtree of the graph while the rest of the DAG is still running. We have a DAG that might run for longer than 24 hours and we’ve encountered a common situation where one node in the graph fails for exogenous reasons. We’d like to fix the exogenous problem and then re-run that node so the dependent nodes can move on, but not interrupt the already running nodes.

Hey, I setup Prefect on fargate EKS on AWS. Everything has been running fine but now the completed pods are not deleted and are skyrocketing the AWS costs. I see a bunch of pods when i run kubectl get pods, (i deleted them manually for now). Any guidance on things I can look at to make sure that something is still not running and AWS is not going to keep racking up the bill?

Hi 👋 I started playing around with Prefect today as an orchestration tool for our Databricks jobs. I've managed to trigger jobs successfully which is great but, I cannot figure out how to run tasks in sequence using the Functional API. I've been browsing through the docs but did not this example with Task Classes. Here's my flow:

with Flow("application-etl") as flow:
    conn = PrefectSecret("DATABRICKS_CONNECTION_STRING_PRE")

    bronze = DatabricksRunNow(job_id=BRONZE_APPLICATION_JOB_ID, name="bronze")

    silver = DatabricksRunNow(job_id=SILVER_APPLICATION_JOB_ID, name="silver")
    silver.set_upstream(bronze)

    gold = DatabricksRunNow(job_id=GOLD_APPLICATION_JOB_ID, name="gold")
    gold.set_upstream(silver)

    bronze(databricks_conn_secret=conn)
    silver(databricks_conn_secret=conn)
    gold(databricks_conn_secret=conn)

flow.run()

I want to run my tasks in the sequence

bronze > silver > gold

. While this executes gold or silver are being started before bronze. I don't know if I'm misunderstanding how

set_upstream

works 🤔 Thanks in advance

Prefect logging question: When I ran below from Mac, both info and debug rows (2) were displayed, but when I registered and ran flow from Cloud, only 1 row, the logger.info row is shown, please advise logger=prefect.context.get("logger") logger.setLevel(logging.DEBUG) logger.info(‘this is test info ) logger.debug('this is test debug')

Hi all. Wondering if anybody have experienced running Prefect inside an API context using a third party worker, like Celery or something. What I mean is, declaring the flow/tasks with Prefect Core, and then calling

flow.run()

within a Celery task for example. Anyone?

Is there a way to create flows in prefect 2.0 from a class function?

from prefect import flow

class MyClass:
    def __init__(self, ct: int=0):
        self.count = ct
        
    @flow
    def runit(self):
        print(f"I have been called {self.count} times")
        self.count += 1
        
        
c = MyClass()
c.runit()

I get the error

TypeError: missing a required argument: 'self'

Is that expected, or a potential bug in orion?

I could of course to

c.runit(c)

but that seems pretty wrong

Hello Team Prefect, I have some question about ECSAgent, because when i tried to run my flow i’ve got botocore.errorfactory.ClientException: An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Container.image should not be null or empty. After having debugged, I’ve detected that the problem was the containerDefinitions used by the agent has two entries, one with my ECSRun configuration and other with only {“name”:“flow”}. But my configuration only had one. My question is, this is a bug or is mandatory using ‘flow’ in the containerDefinitions name parameter?. (I have using prefect 1.2.0, DockerStorage, ECSRun and ECSAgent. And my configuration for the task_definition is directly in the code.)

Hi! I’m currently orchestrating dbt via Prefect. The usual dbt flow uses pygit2 to pull from our main branch, then run the dbt models and create tables into our production schema in Snowflake. I want to create a tool where we can run 2 dbt instances in parallel, one that pulls from the main branch, one that pulls from a dev branch (chosen by a user via parameters). Each of those would create tables in a separate schema in Snowflake. The purpose is to create 2 versions of certain tables to compare and see the effect of the dev branch. Now, my question is what would be the best way to design the flow(s) to this? I’m not sure if I should create 1 flow and some how pull 2 separate branches and have dbt run those 2 in parallel. Or create 1 parent flow that passes the user inputed parameter into the children flow, then each child flow runs a dbt instance? Or something completely different? Sorry for the lengthy question!

Hey! I have been using prefect for a bit now on AKS in Azure for hosting our agents. On the whole this seems to work well, it creates K8s jobs for our flow runs, but regardless of how we trigger them, it only ever seems to to run the amount of concurrent tasks in the flow to be the amount of virtual or physical CPUs on the cluster. I was under the impression if we set the scheduler to "threads" and the num_workers to x2-4 the number of CPUs, the flow tasks would look to create that amount of threads to run the flow? I am comfortable we can scale these as we are simply submitting jobs to a databricks cluster, so we have no CPU bound work at all happening on the cluster, it is purely IO monitoring. Any help would be appreciated, I have attached a sample of how we trigger our flow in the thread. Thanks in advance for any help which could be offered.

Hi, we have a requirement for triggering a flow in one tenant A after another flow running in tenant B completes. this doesn't seem to be how prefect is designed to work (?), but other requirements have made this tenant architecture necessary. is there any official support for this pattern, or will I have to write my own connector? I have looked at taking from the

wait_for_flow_run

method for creating flow of flows, but this seems designed to work with one tenant

Hi, I have a flow of flows where I'm launching two instances of the same flow with different parameters. This pipeline runs every day, but occasionally (about once every two weeks or so), one fails to launch with the following error:

An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family.

Is a reliable way to avoid this or a best practice for launching multiple versions of a flow?

Using the LocalDaskExecutor... is there a way to limit the number of tasks to a specific number? With the schematic that a flow shows based on dependancies, We need to limit the number of executions in the entire flow to only be running 20 tasks at a time. Is there a configuration that Im not seeing?

Hi, do jobs sent to a local agent run in the terminal the agent is listening from? Meaning the job can access a cli authentication and environment variables?

Hey! I try to interact with the GraphQL api and I want to set the default parameters for a flow group. What is the right way to pass the parameters (as dict/json) into the query?

Hi Good morning, is anyone familiar with dropping the following labels from a job/task in a kubernetes agent:

<http://prefect.io/flow_id=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxx|prefect.io/flow_id=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxx>                                                                                                                                                            
<http://prefect.io/flow_run_id=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxx|prefect.io/flow_run_id=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxx>                                                                                                                                                         <http://prefect.io/identifier=xxxxxxxx|prefect.io/identifier=xxxxxxxx>

I’d like to use a Secret stored in Prefect Cloud in my run config. I have stored the secret using

>>> from prefect import Client
>>> client = Client()
>>> client.set_secret(name="test", value="test")
>>> client.set_secret(name="AWS_ACCOUNT_ID-dev", value="123456789098")

It seems like I should be able to use

from prefect.client import Secret

aws_account_id = Secret('AWS_ACCOUNT_ID-' + RUN_ENV).get()

However, I receive the following error:

ValueError: Local Secret "AWS_ACCOUNT_ID-dev" was not found.

How can I retrieve a secret value outside of a task?

Hi, which python path is used for

LocalRun

?? Can I use other virtual environments’ python path like virtual envs in pyenv or conda path?