Hi, when running on ECSRun and setting env variables they are auto casting strings to ints which causes boto3 exception, is there a way to avoid auto casting?

Hi folks, if I return a dictionary from a task, is there a way to unpack that result into

kwargs

for the next task downstream? Something like;

# returns {"path": "some/restapi/path", "start_time": datetime...}
tasks = generate_tasks(first_set)

# Needs a path and start_time argument
histories = fetch_history.map(**tasks)

But this gives me the error

expression after ** must be a mapping with a "str" key

Hi folks, was just looking at prefect example like this.. However, let's say instead of running the steps as python code, i have some cli commands.

with Flow("NLP") as nlp_flow:
    aws s3 cp blah blah
    sometool /data/a.csv

Is it possible to add the steps as cli commands instead ?

Hi everyone. I have had encountered this error for weeks, and it keeps getting often. Prefect cloud always showing 404 on Flow pages, only to certain flow, but now it's almost for all new Flow registered I took a look on the network tools, after it got 404, it keeps hitting API and gets 200 in return, until forever. How to resolve this?

Hi guys, I'm looking to build a

class MyFlowRunner(prefect.engine.flow_runner.FlowRunner)

(or subclassing another component - open to suggestions!) inside a custom package

my-prefect

to send some HTTP requests before a

Flow

starts, and after it ends. My questions are: • where would be the best way for me to implement such a functionality? along with the considerations: • keep the Prefect API to be as pure as possible • allow the implementation of the

MyFlowRunner

(or some other custom component) to be uncoupled, i.e. if a project upgrades

my-prefect

, the flow configuration doesn't have to change at all Would appreciate comments/suggestions/discussions from whoever are interested/have had experience doing so!

Hi Prefect, why is this an invalid schedule in the UI? As you can see it is valid cron syntax 🤷

Quick question. I know that .map() can be used as a single for-loop in a flow. Is there a way to use .map() to create a nested for-loop though?

Quick question from a newbie in prefect. What is the easiest way to set up prefect and the agents to distribute flows to different machines in the local network (see picture). Maybe you can give me some advice and share some readings regarding this. Thanks

Hey Prefect community, does anyone know some common causes for flow runs to get stuck in the ‘Submitted’ state? We’re running a docker agent in DEV & TEST, both are deployed from the same configuration, DEV works fine, but all flows to TEST get stuck in ‘Submitted’.

I am trying to import pandas. I have prefect cloud, AKS agent. I have modified the yaml file for pod to include pip install pandas but when I run the flow it can't find the pandas package. When I log onto the pod where the agent is I see that it is installed when the pod gets created by Kubernetes.When I run the flow I get: Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n ModuleNotFoundError("No module named \'pandas\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

Please help me to clear up my confusion on instantiating prefect flows. I have three dbt commands to run:

dbt deps
dbt snapshot
dbt run

For now I'm running them by reusing the instance of DbtShellTask:

dbt_task = DbtShellTask(name="Running dbt commands", ...)

dbt_clean_task_invoke   = dbt_task("dbt deps")
dbt_deps_task_invoke    = dbt_task("dbt snapshot")
dbt_run_task_invoke     = dbt_task("dbt run")

With having this setup, my schematic will have three boxes(tasks) with name

Running dbt commands

which is difficult to identify which is which. In order to fix that, do I have to instantiate for each command like below.

dbt_clean_task = DbtShellTask(name="Running dbt deps command", ...)
dbt_deps_task  = DbtShellTask(name="Running dbt snapshot command", ...)
dbt_run_task   = DbtShellTask(name="Running dbt run command", ...)

dbt_clean_task_invoke   = dbt_clean_task("dbt deps")
dbt_deps_task_invoke    = dbt_deps_task("dbt snapshot")
dbt_run_task_invoke     = dbt_run_task("dbt run")

Does someone know how to solve this issue? I can't log in since yesterday, but it is possible to reset the password.

Hi all! Is there a way I can run a flow with a custom CloudFlowRunner without having to modify the agent code?

Is there a way to create a task that's reused with different parameters, that for each use is given a different name that then appears in the schematic visualization? I thought that doing

@task(task_run_name="{name_val}")

and then passing in a name_val variable would do it, but in the dashboard everything is still showing up as the name of the function definition

Hi! I've been using Prefect for a while and I'm investigating how to scale it. My organization uses Azure and I was looking for something similar to ECS+Fargate (ECSAgent) integration but in Azure. The closest thing to Fargate is Azure Container Instances. I don't want to use Kubernetes because I want to pay per task, not to have a cluster almost always on idle. Is it possible that it is going to be supported in the future (aka ACIAgent)? Thank you very much in advance

Hi, my company have recently tested Prefect - we love it and we are thinking of switching from Airflow! The caveat is we need fully on-prem deployment. Is there an estimated cost (order of magnitude will do) for the enterprise option?

What would be the best structure for when I have a task that, on success should just return data normally that is appended to a list, but on failure should run another task that gets data from another source, and appends that instead? Can you use "with case" blocks on a tasks state, rather than on a tasks output? Basically, if the original task fails, I want to replace it's output with the output of another task/function instead, which is then appended the same way

Question about mapped tasks and custom state handlers for them - I've got a state handler for when a mapped task fails, however because the object passed to the state handler is a Task and not a TaskRun, the task.name is the generic task name, and not the mapped task name. Similarly in the state handler I don't have the task_run_id either. Is there any way to get the task_run_name in the state handler for a mapped task? Forgot to mention - if I try

object.task_run_name

it prints the format string used to generate the mapped task name, not the actual task name. I'm using the pattern here where my

task_run_name = "{table_name}"

- and I get

{table_name}

output in the state handler, because that's what it's set as at the Task level, rather than it being computed at the TaskRun level

Hey folks, Looking for some assistance here if possible, as I’m a little confused. Given the flow below, which basically orchestrates a bunch of other flows, I’d expect this to run the dependant flows as per the dependencies defined, but it does not, it kicks them all of at the same time. * see thread It was my understanding given the schematic that is drawn that, that it would wait for the upstream tasks to be completed (maybe my upstream vs down stream is confused here). The ‘FlowRunner’ class here is just a wrapper around an environment variable to run locally or via Kubernetes. * see thread Can anyone see any obvious mistakes here.

What would be the approach to conditionally register a flow depending on both the flow itself and the tasks used by the flow? So, the

idempotency_key

used when calling

flow.register()

would take into account both flow and task changes. From what I can tell,

flow.serialized_hash()

only changes if changes are made to the flow's metadata. It does not take into account if a task within a flow changes.

Hi community, how can I pass parameters to a ShellTask? I can't seem to find an example for it in docs

Hey community, is there a way to loop over a group of tasks in sequential order ? Looping ironically doesnt seem to do this. I dont want to transfer any data between loops, just kind of want to hack a flow to run multiple times within the minute

Hello, How could we disable flow run retry when this below happen?

No heartbeat detected from the remote task; retrying the run.This will be retry 1 of 2.

In most situation, we do want to retry the run. But we have a particular flow that we actually don’t want any retrying when the run lost the heartbeat. we just it to fail. How could we do that?

Hey community, Trying to write a graphql query that I can specify the

project

and the

flow name

and get a list of currently active flows and their state ( running, or complete etc ). Is this possible ? Cant seem to find anything for the use case...

Hi everyone, I noticed that when we’re submitting flow run [create_flow_run] we got very long response times, usually several seconds [about 5s] sometimes upto 15sec? is that expected behaviour?

Hey all! Wondering if anyone knows if Prefect has any prometheus metrics exposed. We’re trying to figure out how to integrate our current AlertManager monitoring with prefect server and its flows. Have other people solved monitoring in an elegant way?

Hi community, I notice the schedule flow runs come with funny names (e.g. melodic-donkey). Is it possible to template the flow run names? Looked through the docs but it's not mentioned anywhere.

Hey, I'm new to prefect and have a question regarding sharing objects between different flows. I have two different flows,

flow_1

initializes the object and

flow_2

uses that object, but I'm not able to achieve this. This is the example code not exact but similar to what I'm trying to achieve. adding code to thread

Hi there! Using Prefect for a few weeks now on one of my consulting jobs (in prod). Just started to get into using Prefect Server and have a few questions - that’s why I’m here 🙂

Can anyone tell me what

KeyError: 110

is? I think it is a Prefect error, though it may be Dask. The stack trace has it coming out of the executor.py code. (trace to follow in thread)