Hi folks! We want to run an incremental flow in Prefect. I.e., we move some data from Snowflake to a AWS SageMaker feature store. We have this part working. However, what we want to do is run the flow on a regular schedule, and each time, only move the rows in Snowflake that are not already moved to AWS from the previous run. Is there a good way to schedule incremental runs like this in Prefect itself? If someone knows of an example to point us to of how this is done, we'd be grateful 🙂

Are flow objects mostly similar or fundamentally different in Prefect Core vs Prefect Orion? I’d be specifically interested in why in Prefect Core you decided to make Flow a context manager and not a decorator like it is now in Prefect Orion. Was that a more or less arbitrary choice, or is there a deeper reason why in the old API a decorator wasn’t possible or desirable? There’s the statement at https://www.prefect.io/blog/announcing-prefect-orion/ that now flows aren’t DAGs any more. But I couldn’t find an explanation what they are now. I presume the prefect.flow decorator still does some analysis and generates a DAG-like data structure that represents task dependencies? Is this change in what a flow object is related to the change context manager -> decorator? https://docs.prefect.io/core/concepts/flows.html https://orion-docs.prefect.io/api-ref/prefect/flows/

I've got a case where I start a local flow with num_workers in a LocalDaskExecutor set to 6, but the task seems to start many more tasks than that. For example, that flow is currently showing 15 tasks running simultaneously. Is there something else needed to limit the number of simultaneous tasks? It overloads the machine as is.

hi there,

task_run_name

when templated with input names can be very helpful in debugging mapped tasks. Are they only available on Prefect backend (Cloud and Server) and not Core? If I'm doing local debugging with

flow_result = flow.run()

, let's say I have a flow with

task_a

mapped 1000 times and

task_a[25]

failed but the other indices succeed. What's the quickest way to find out which input caused it to fail? I don't think I can access the result of

flow_result.result[task_a].result[25]

Hi everyone, I’m looking for suggestions for how to use prefect together with a Spark cluster running on aws. Is there an equivalent to Airflow’s SparkSubmitOperator ? I’ve only found a DataBricksTask in the docs, but I’m not interested in getting a Databricks subscription at this point.

what's the proper way to make a Kubernetes agent (and a Kubernetes Run) use an S3 storage for the flows? we tried the k8s

service account

but it seems insufficient and we noticed some warning in the docs about that but couldn't really decipher what it would mean for us since we don't use these methods to define permissions:

I'm on "account settings" -> "account" and I don't see a way to see my past payment activity. Where do I see a recap of each credit card charge?

Hi everyone, I am having issues with one task in my flow and not sure why. I see this on the logs when the task the run. Other tasks run find and I have tried a different flow on the same agent with this error. I have also tried to run the flow as a process with flow.run() and it completes successfully

Hi everyone! First of all happy new year everybody 🙂 I feel that I don’t fully understand the Docker Storage. As far as I know, you need to serialize the flow (using cloudpickle) and when the prefect agent requests a flow run, it downloads the flow and run it. However, if you are using the Docker Storage I don’t understand why cloudpickle is still used, as I expect that the whole flow is pushed as a docker image to some repository. My question comes from a project where I use docker storage (the python 3.8 image

prefecthq/prefect:0.15.11-python3.8

), but I am using python 3.9 to register the flow to prefect cloud. It fails because I am using different python versions. Could you please clarify me how it works? Thank you very much!

Hey all, happy new year! I have an odd failure I'm having trouble understanding. I have a flow with a single parameter input that runs on a Fargate cluster with a Dask executor. With some parameters it runs prefectly 😉 While with others, I receive the following error:

botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: TaskDefinition is inactive

after this:

Creating a new Dask cluster with `dask_cloudprovider.aws.ecs.FargateCluster`...

In the run config, both when it succeeds and when it fails the task_definition is null, and all the other settings are the same, has anybody encountered a similar issue and has any idea what could go wrong?

Hi! is there a option in prefect something similiar to smart sensor in airflow?

Hi everyone, we are cloud users and are getting a 'Usage Exceeded' message in our agent logs as of last night ~9pm GMT. Could this be some kind of new year related issue? I don't remember hearing about a hard limit on runs but that if our usage was higher than planned we'd get a notification to discuss new limits/rates?

Hi everyone! I see that there is an option “max_scheduled_runs_per_flow” in prefect server that was added in August but what about PrefectHQ/prefect? When it will be added there?

Hi, is there a possibility to scale prefect ECS-agents (auto scaling) and if so can someone point me in the right direction, Im struggling to understand how a new agent would register with prefect (and with existing flows)

Hello prefect community 🙂, We're using prefect with K8S. Sometime some tasks fail because the pods that run them get killed (for OOM or anything else, doesn't matter), which leads to a "pod missed heartbeat..." message. The issue is that when a flow/task fails like this, the retry mechanism and the slack handler doesn't work. Is there a way to retry tasks on this kind of failure or/and to get any notification about the status ?

good morning and happy new year! Is it possible to have a task that runs if: 1. some upstream tasks are successful AND 2. some upstream tasks are successful or failed (just completed)

When testing flows locally, because of breadth first execution, my flow's mapped task is storing a large number of high-memory objects at the same time that cause memory error/segmentation fault. Is there a way to fix this?

Happy New Year everyone! Is there a way, from within an active Flow, to know whether the flow was triggered by cron vs Quick run vs Run?

Is there a way to set terminal task for the flow after its creation? I see that the flow object has

set_reference_task()

but not

set_terminal_task()

. I'm trying to debug by stopping the flow early and inspect the results

Happy New Year everyone! When I run a GE checkpoint in GE it works, but in prefect task, I'm getting the error:

TypeError: 'Checkpoint' object is not subscriptable

Hey everyone - I'm a relatively new user and am struggling with my development workflow. I have found good introductory resources for running both locally and using prefect cloud but what I am struggling with is how to develop/debug locally and deploy to the cloud. For some context, we currently just have 1 flow that will simply be executing dbt commands. We are deploying to azure and using docker. I have a task that clones our dbt repo from github before executing the commands. We also have an action to register our flow each time the flow is updated in github. Can anyone point me in the direction of best practices for developing and testing locally while using cloud for production?

how do i view previous versions of a flow and see their creation date and by whom? https://docs.prefect.io/orchestration/concepts/flows.html#ui

Parameters and Case statements are appearing to be less intuitive than I had thought…I can’t figure out a simple way to do something like the following:

val = Parameter("param", False):

case(val, True):
    do_stuff()

case(val, False):
    dont_do_stuff()

If my paramter is an actual value, i.e. “value”, it skips both because it’s neither True nor False. So I’m left to make another task to check the value. Is this correct?

Hi all. I'm trying to set up a Prefect docker agent as a systemd service (CentOS). What is the recommended practice for storing the prefect API Key and passing it on to the systemd service?

With Orion, will subflows be mappable so we can run a dynamic amount of subflows based on some task result? Didn’t find any previous details on this

Hi, I found something interesting about Cancelling and Failing a run. Whenever I press Cancel in my Flow run, all tasks should be stopped and have the

state.Cancelling

and

state.Cancelled

as

True

, but none of them occurs. It only register as

Cancelling

or

Cancelled

when I put the flow to the State (UI) as Cancelled. Another weird fact is if I put the State as Cancelled, my tasks keeps running until it finishes (either in fail or succeed). I'm currently working on State handlers, how can I be sure that I'm cancelling my flow runs and consequently my tasks for that flow run?

I have a problem where a Flow Run went from

Scheduled

to

Cancelled

without pressing cancel in the UI. By the looks of the following query the Flow Run was cancelled because a new version of the Flow was released (which archived the old flow). Is this expected behavior?

Hi all, thanks in advance for any pointers. Does prefect have an API I could hit from Github? Specifically, I want a commit to our github repo to start a new run of a prefect flow. I was able to find Prefect webhooks for doing things after a flow runs, but I am trying to find a way to make a github webhook invoke a prefect flow run, instead. Any pointers?

Hi, can I trust the order of mapped tasks? As in if I have a list

l1 = [a, b, c]

and I generate

l2 = task_name.map(l1)

, can I trust that element 0 in

l2

derives from element 0 from

l1

?

We're running all our flows via ECSRun to kick off a Fargate task to run the flow. It all works fine but what I'm seeing are a lot of Fargate zombie tasks still running after the flow has finished. I need to investigate this a little more as it may be failed flows and/or flows we cancel via the Prefect console... not sure at this stage. Has anyone else had a similar issue? (a quick search of this channel didn't reveal much). Many thanks!